[management] incremental network map builder (#4753)

Author: crn4

go.mod

@@ -56,6 +56,7 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/hashicorp/go-secure-stdlib/base62 v0.1.2
 	github.com/hashicorp/go-version v1.6.0
+	github.com/jackc/pgx/v5 v5.5.5
 	github.com/libdns/route53 v1.5.0
 	github.com/libp2p/go-netroute v0.2.1
 	github.com/mdlayher/socket v0.5.1
@@ -183,7 +184,6 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/jackc/pgpassfile v1.0.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
-	github.com/jackc/pgx/v5 v5.5.5 // indirect
 	github.com/jackc/puddle/v2 v2.2.1 // indirect
 	github.com/jeandeaual/go-locale v0.0.0-20240223122105-ce5225dcaa49 // indirect
 	github.com/jinzhu/inflection v1.0.0 // indirect

management/main.go

@@ -1,11 +1,19 @@
 package main
 
 import (
-	"github.com/netbirdio/netbird/management/cmd"
+	"log"
+	"net/http"
+	// nolint:gosec
+	_ "net/http/pprof"
 	"os"
+
+	"github.com/netbirdio/netbird/management/cmd"
 )
 
 func main() {
+	go func() {
+		log.Println(http.ListenAndServe("localhost:6060", nil))
+	}()
 	if err := cmd.Execute(); err != nil {
 		os.Exit(1)
 	}

management/server/account.go

@@ -53,6 +53,9 @@ const (
 	peerSchedulerRetryInterval = 3 * time.Second
 	emptyUserID                = "empty user ID in claims"
 	errorGettingDomainAccIDFmt = "error getting account ID by private domain: %v"
+
+	envNewNetworkMapBuilder  = "NB_EXPERIMENT_NETWORK_MAP"
+	envNewNetworkMapAccounts = "NB_EXPERIMENT_NETWORK_MAP_ACCOUNTS"
 )
 
 type userLoggedInOnce bool
@@ -109,6 +112,11 @@ type DefaultAccountManager struct {
 	loginFilter *loginFilter
 
 	disableDefaultPolicy bool
+
+	holder *types.Holder
+
+	expNewNetworkMap     bool
+	expNewNetworkMapAIDs map[string]struct{}
 }
 
 func isUniqueConstraintError(err error) bool {
@@ -196,6 +204,18 @@ func BuildManager(
 		log.WithContext(ctx).Debugf("took %v to instantiate account manager", time.Since(start))
 	}()
 
+	newNetworkMapBuilder, err := strconv.ParseBool(os.Getenv(envNewNetworkMapBuilder))
+	if err != nil {
+		log.WithContext(ctx).Warnf("failed to parse %s, using default value false: %v", envNewNetworkMapBuilder, err)
+		newNetworkMapBuilder = false
+	}
+
+	ids := strings.Split(os.Getenv(envNewNetworkMapAccounts), ",")
+	expIDs := make(map[string]struct{}, len(ids))
+	for _, id := range ids {
+		expIDs[id] = struct{}{}
+	}
+
 	am := &DefaultAccountManager{
 		Store:                    store,
 		geo:                      geo,
@@ -217,6 +237,10 @@ func BuildManager(
 		permissionsManager:       permissionsManager,
 		loginFilter:              newLoginFilter(),
 		disableDefaultPolicy:     disableDefaultPolicy,
+		holder:                   types.NewHolder(),
+
+		expNewNetworkMap:     newNetworkMapBuilder,
+		expNewNetworkMapAIDs: expIDs,
 	}
 
 	am.startWarmup(ctx)
@@ -395,6 +419,9 @@ func (am *DefaultAccountManager) UpdateAccountSettings(ctx context.Context, acco
 	}
 
 	if updateAccountPeers || extraSettingsChanged || groupChangesAffectPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return nil, err
+		}
 		go am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -1477,6 +1504,10 @@ func (am *DefaultAccountManager) SyncUserJWTGroups(ctx context.Context, userAuth
 		}
 
 		if removedGroupAffectsPeers || newGroupsAffectsPeers {
+			if err := am.RecalculateNetworkMapCache(ctx, userAuth.AccountId); err != nil {
+				return err
+			}
+
 			log.WithContext(ctx).Tracef("user %s: JWT group membership changed, updating account peers", userAuth.UserId)
 			am.BufferUpdateAccountPeers(ctx, userAuth.AccountId)
 		}
@@ -2129,6 +2160,11 @@ func (am *DefaultAccountManager) UpdatePeerIP(ctx context.Context, accountID, us
 	}
 
 	if updateNetworkMap {
+		peer, err := am.Store.GetPeerByID(ctx, store.LockingStrengthNone, accountID, peerID)
+		if err != nil {
+			return err
+		}
+		am.updatePeerInNetworkMapCache(peer.AccountID, peer)
 		am.BufferUpdateAccountPeers(ctx, accountID)
 	}
 	return nil

management/server/account/manager.go

@@ -128,4 +128,5 @@ type Manager interface {
 	GetCurrentUserInfo(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)
 	SetEphemeralManager(em ephemeral.Manager)
 	AllowSync(string, uint64) bool
+	RecalculateNetworkMapCache(ctx context.Context, accountId string) error
 }

management/server/account_test.go

@@ -1154,7 +1154,16 @@ func TestAccountManager_AddPeerWithUserID(t *testing.T) {
 	assert.Equal(t, peer.IP.String(), fmt.Sprint(ev.Meta["ip"]))
 }
 
+func TestAccountManager_NetworkUpdates_SaveGroup_Experimental(t *testing.T) {
+	t.Setenv(envNewNetworkMapBuilder, "true")
+	testAccountManager_NetworkUpdates_SaveGroup(t)
+}
+
 func TestAccountManager_NetworkUpdates_SaveGroup(t *testing.T) {
+	testAccountManager_NetworkUpdates_SaveGroup(t)
+}
+
+func testAccountManager_NetworkUpdates_SaveGroup(t *testing.T) {
 	manager, account, peer1, peer2, peer3 := setupNetworkMapTest(t)
 
 	group := types.Group{
@@ -1205,7 +1214,16 @@ func TestAccountManager_NetworkUpdates_SaveGroup(t *testing.T) {
 	wg.Wait()
 }
 
+func TestAccountManager_NetworkUpdates_DeletePolicy_Experimental(t *testing.T) {
+	t.Setenv(envNewNetworkMapBuilder, "true")
+	testAccountManager_NetworkUpdates_DeletePolicy(t)
+}
+
 func TestAccountManager_NetworkUpdates_DeletePolicy(t *testing.T) {
+	testAccountManager_NetworkUpdates_DeletePolicy(t)
+}
+
+func testAccountManager_NetworkUpdates_DeletePolicy(t *testing.T) {
 	manager, account, peer1, _, _ := setupNetworkMapTest(t)
 
 	updMsg := manager.peersUpdateManager.CreateChannel(context.Background(), peer1.ID)
@@ -1239,7 +1257,16 @@ func TestAccountManager_NetworkUpdates_DeletePolicy(t *testing.T) {
 	wg.Wait()
 }
 
+func TestAccountManager_NetworkUpdates_SavePolicy_Experimental(t *testing.T) {
+	t.Setenv(envNewNetworkMapBuilder, "true")
+	testAccountManager_NetworkUpdates_SavePolicy(t)
+}
+
 func TestAccountManager_NetworkUpdates_SavePolicy(t *testing.T) {
+	testAccountManager_NetworkUpdates_SavePolicy(t)
+}
+
+func testAccountManager_NetworkUpdates_SavePolicy(t *testing.T) {
 	manager, account, peer1, peer2, _ := setupNetworkMapTest(t)
 
 	group := types.Group{
@@ -1288,7 +1315,16 @@ func TestAccountManager_NetworkUpdates_SavePolicy(t *testing.T) {
 	wg.Wait()
 }
 
+func TestAccountManager_NetworkUpdates_DeletePeer_Experimental(t *testing.T) {
+	t.Setenv(envNewNetworkMapBuilder, "true")
+	testAccountManager_NetworkUpdates_DeletePeer(t)
+}
+
 func TestAccountManager_NetworkUpdates_DeletePeer(t *testing.T) {
+	testAccountManager_NetworkUpdates_DeletePeer(t)
+}
+
+func testAccountManager_NetworkUpdates_DeletePeer(t *testing.T) {
 	manager, account, peer1, _, peer3 := setupNetworkMapTest(t)
 
 	group := types.Group{
@@ -1341,7 +1377,16 @@ func TestAccountManager_NetworkUpdates_DeletePeer(t *testing.T) {
 	wg.Wait()
 }
 
+func TestAccountManager_NetworkUpdates_DeleteGroup_Experimental(t *testing.T) {
+	t.Setenv(envNewNetworkMapBuilder, "true")
+	testAccountManager_NetworkUpdates_DeleteGroup(t)
+}
+
 func TestAccountManager_NetworkUpdates_DeleteGroup(t *testing.T) {
+	testAccountManager_NetworkUpdates_DeleteGroup(t)
+}
+
+func testAccountManager_NetworkUpdates_DeleteGroup(t *testing.T) {
 	manager, account, peer1, peer2, peer3 := setupNetworkMapTest(t)
 
 	updMsg := manager.peersUpdateManager.CreateChannel(context.Background(), peer1.ID)
@@ -1377,6 +1422,14 @@ func TestAccountManager_NetworkUpdates_DeleteGroup(t *testing.T) {
 		return
 	}
 
+	for drained := false; !drained; {
+		select {
+		case <-updMsg:
+		default:
+			drained = true
+		}
+	}
+
 	wg := sync.WaitGroup{}
 	wg.Add(1)
 	go func() {
@@ -1736,7 +1789,9 @@ func TestAccount_Copy(t *testing.T) {
 				Address:   "172.12.6.1/24",
 			},
 		},
+		NetworkMapCache: &types.NetworkMapBuilder{},
 	}
+	account.InitOnce()
 	err := hasNilField(account)
 	if err != nil {
 		t.Fatal(err)

management/server/dns.go

@@ -117,6 +117,9 @@ func (am *DefaultAccountManager) SaveDNSSettings(ctx context.Context, accountID
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 

management/server/group.go

@@ -114,6 +114,9 @@ func (am *DefaultAccountManager) CreateGroup(ctx context.Context, accountID, use
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -182,6 +185,9 @@ func (am *DefaultAccountManager) UpdateGroup(ctx context.Context, accountID, use
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -250,6 +256,9 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -318,6 +327,9 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -481,6 +493,9 @@ func (am *DefaultAccountManager) GroupAddPeer(ctx context.Context, accountID, gr
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -519,6 +534,9 @@ func (am *DefaultAccountManager) GroupAddResource(ctx context.Context, accountID
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -547,6 +565,9 @@ func (am *DefaultAccountManager) GroupDeletePeer(ctx context.Context, accountID,
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}
 
@@ -585,6 +606,9 @@ func (am *DefaultAccountManager) GroupDeleteResource(ctx context.Context, accoun
 	}
 
 	if updateAccountPeers {
+		if err := am.RecalculateNetworkMapCache(ctx, accountID); err != nil {
+			return err
+		}
 		am.UpdateAccountPeers(ctx, accountID)
 	}