sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-07-17 20:19:55 +00:00
Code Issues Packages Projects Releases Wiki Activity

expand network managers

Browse Source
This commit is contained in:
pascal
2026-06-05 15:15:04 +02:00
parent a2b0399212
commit 14f9754db7
6 changed files with 172 additions and 85 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
management/server/affectedpeers/resolver.go
View File

@@ -24,6 +24,7 @@ import (
rpservice "github.com/netbirdio/netbird/management/internals/modules/reverseproxy/service"
resourceTypes "github.com/netbirdio/netbird/management/server/networks/resources/types"
routerTypes "github.com/netbirdio/netbird/management/server/networks/routers/types"
networkTypes "github.com/netbirdio/netbird/management/server/networks/types"
"github.com/netbirdio/netbird/management/server/store"
"github.com/netbirdio/netbird/management/server/types"
"github.com/netbirdio/netbird/route"
@@ -60,8 +61,11 @@ func Load(ctx context.Context, s store.Store, accountID string, c Change) (*Snap
return snap, nil
}
hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0
needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0
// Changed resources contribute their group IDs to the changed-group set during
// the walk (see collectFromExplicitResources), so they drive the group walkers.
hasGroupOrPeerChange := len(c.ChangedGroupIDs) > 0 || len(c.ChangedPeerIDs) > 0 || len(c.Resources) > 0
hasNetworkObject := len(c.ResourceIDs) > 0 || len(c.NetworkIDs) > 0 || len(c.Routers) > 0 || len(c.Resources) > 0 || len(c.Networks) > 0
needsPolicies := hasGroupOrPeerChange || len(c.PostureCheckIDs) > 0 || len(c.Policies) > 0 || hasNetworkObject
needsRoutersResources := needsPolicies // the resource<->router bridge can fire whenever policies/resources/networks are in play
var err error
@@ -127,6 +131,9 @@ type Change struct {
ChangedPeerIDs []string
Policies []*types.Policy
Routes []*route.Route
Routers []*routerTypes.NetworkRouter
Resources []*resourceTypes.NetworkResource
Networks []*networkTypes.Network
PostureCheckIDs []string
ResourceIDs []string
NetworkIDs []string
@@ -147,6 +154,9 @@ func (c Change) isEmpty() bool {
len(c.ChangedPeerIDs) == 0 &&
len(c.Policies) == 0 &&
len(c.Routes) == 0 &&
len(c.Routers) == 0 &&
len(c.Resources) == 0 &&
len(c.Networks) == 0 &&
len(c.PostureCheckIDs) == 0 &&
len(c.ResourceIDs) == 0 &&
len(c.NetworkIDs) == 0 &&
@@ -165,8 +175,8 @@ func (snap *Snapshot) Expand(ctx context.Context, accountID string, c Change) []
return nil
}
r := newResolver(ctx, snap, accountID, c)
log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d postureChecks=%v resources=%v networks=%v",
accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs)
log.WithContext(ctx).Tracef("affectedpeers expand start: account=%s changedGroups=%v changedPeers=%v policies=%d routes=%d routers=%d resources=%d networks=%d postureChecks=%v resourceIDs=%v networkIDs=%v",
accountID, c.ChangedGroupIDs, c.ChangedPeerIDs, len(c.Policies), len(c.Routes), len(c.Routers), len(c.Resources), len(c.Networks), c.PostureCheckIDs, c.ResourceIDs, c.NetworkIDs)
r.walk()
return r.expand()
}
@@ -222,6 +232,9 @@ func newResolver(ctx context.Context, snap *Snapshot, accountID string, c Change
func (r *resolver) walk() {
r.collectFromExplicitPolicies()
r.collectFromExplicitRoutes(r.change.Routes)
r.collectFromExplicitRouters(r.change.Routers)
r.collectFromExplicitResources(r.change.Resources)
r.collectFromExplicitNetworks(r.change.Networks)
r.collectFromPostureChecks(r.change.PostureCheckIDs)
if len(r.changedGroupSet) > 0 || len(r.changedPeerSet) > 0 {
@@ -339,6 +352,62 @@ func (r *resolver) collectFromExplicitRoutes(routes []*route.Route) {
}
}
// collectFromExplicitRouters folds the routing peers carried by changed router
// objects (old and/or new state) directly, and marks their networks so the
// router<->source bridge folds the source peers of policies serving them. Carrying
// the old router object here is how a repointed router's previous routing peers
// stay affected without a post-commit read.
func (r *resolver) collectFromExplicitRouters(routers []*routerTypes.NetworkRouter) {
for _, router := range routers {
if router == nil {
continue
}
log.WithContext(r.ctx).Tracef("collectFromExplicitRouters: changed router %s on network %s -> folding peerGroups=%v peer=%q and marking network for source bridge",
router.ID, router.NetworkID, router.PeerGroups, router.Peer)
addAll(r.groupSet, router.PeerGroups)
if router.Peer != "" {
r.peerSet[router.Peer] = struct{}{}
}
if router.NetworkID != "" {
r.networkIDs[router.NetworkID] = struct{}{}
}
}
}
// collectFromExplicitResources marks the networks of changed resource objects
// (old and/or new state) so the bridge folds their routers and the source peers
// of policies targeting them, and treats each resource's group IDs as changed
// groups so policies targeting the resource through a now-detached (old) group —
// which the post-update group.Resources no longer links — still refresh.
func (r *resolver) collectFromExplicitResources(resources []*resourceTypes.NetworkResource) {
for _, resource := range resources {
if resource == nil {
continue
}
log.WithContext(r.ctx).Tracef("collectFromExplicitResources: changed resource %s on network %s -> marking network for bridge and treating groups %v as changed",
resource.ID, resource.NetworkID, resource.GroupIDs)
addAll(r.changedGroupSet, resource.GroupIDs)
if resource.NetworkID != "" {
r.networkIDs[resource.NetworkID] = struct{}{}
}
}
}
// collectFromExplicitNetworks marks changed network objects (old and/or new
// state) so the bridge folds their routers and the source peers of policies
// serving their resources. A network has no groups/peers of its own.
func (r *resolver) collectFromExplicitNetworks(networks []*networkTypes.Network) {
for _, network := range networks {
if network == nil {
continue
}
log.WithContext(r.ctx).Tracef("collectFromExplicitNetworks: changed network %s -> marking for bridge", network.ID)
if network.ID != "" {
r.networkIDs[network.ID] = struct{}{}
}
}
}
func (r *resolver) collectFromPostureChecks(postureCheckIDs []string) {
if len(postureCheckIDs) == 0 {
return

52
management/server/group.go
View File

@@ -221,10 +221,14 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
}
var eventsToStore []func()
var snaps []*affectedpeers.Snapshot
var changes []affectedpeers.Change
var globalErr error
groupIDs := make([]string, 0, len(groups))
createdCount := 0
for _, newGroup := range groups {
change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}}
var snap *affectedpeers.Snapshot
err = am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
if err = validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
return err
@@ -241,34 +245,31 @@ func (am *DefaultAccountManager) CreateGroups(ctx context.Context, accountID, us
return err
}
groupIDs = append(groupIDs, newGroup.ID)
events := am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
eventsToStore = append(eventsToStore, events...)
return nil
snap, err = affectedpeers.Load(ctx, transaction, accountID, change)
return err
})
if err != nil {
log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err)
if len(groupIDs) == 1 {
if createdCount == 0 {
return err
}
globalErr = errors.Join(globalErr, err)
// continue updating other groups
continue
}
createdCount++
snaps = append(snaps, snap)
changes = append(changes, change)
}
for _, storeEvent := range eventsToStore {
storeEvent()
}
affectedPeerIDs := am.ResolveAffectedPeers(ctx, am.Store, accountID, affectedpeers.Change{ChangedGroupIDs: groupIDs})
if len(affectedPeerIDs) > 0 {
log.WithContext(ctx).Debugf("CreateGroups %v: updating %d affected peers: %v", groupIDs, len(affectedPeerIDs), affectedPeerIDs)
am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs)
} else {
log.WithContext(ctx).Tracef("CreateGroups %v: no affected peers", groupIDs)
}
am.dispatchAffected(ctx, accountID, snaps, changes)
return globalErr
}
@@ -287,11 +288,13 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
}
var eventsToStore []func()
var snaps []*affectedpeers.Snapshot
var changes []affectedpeers.Change
var globalErr error
groupIDs := make([]string, 0, len(groups))
for _, newGroup := range groups {
events, err := am.updateSingleGroup(ctx, accountID, userID, newGroup)
change := affectedpeers.Change{ChangedGroupIDs: []string{newGroup.ID}}
events, snap, err := am.updateSingleGroup(ctx, accountID, userID, newGroup, change)
if err != nil {
log.WithContext(ctx).Errorf("failed to update group %s: %v", newGroup.ID, err)
if len(groups) == 1 {
@@ -301,26 +304,22 @@ func (am *DefaultAccountManager) UpdateGroups(ctx context.Context, accountID, us
continue
}
eventsToStore = append(eventsToStore, events...)
groupIDs = append(groupIDs, newGroup.ID)
snaps = append(snaps, snap)
changes = append(changes, change)
}
for _, storeEvent := range eventsToStore {
storeEvent()
}
affectedPeerIDs := am.ResolveAffectedPeers(ctx, am.Store, accountID, affectedpeers.Change{ChangedGroupIDs: groupIDs})
if len(affectedPeerIDs) > 0 {
log.WithContext(ctx).Debugf("UpdateGroups %v: updating %d affected peers: %v", groupIDs, len(affectedPeerIDs), affectedPeerIDs)
am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs)
} else {
log.WithContext(ctx).Tracef("UpdateGroups %v: no affected peers", groupIDs)
}
am.dispatchAffected(ctx, accountID, snaps, changes)
return globalErr
}
func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountID, userID string, newGroup *types.Group) ([]func(), error) {
func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountID, userID string, newGroup *types.Group, change affectedpeers.Change) ([]func(), *affectedpeers.Snapshot, error) {
var events []func()
var snap *affectedpeers.Snapshot
err := am.Store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
if err := validateNewGroup(ctx, transaction, accountID, newGroup); err != nil {
return err
@@ -341,9 +340,12 @@ func (am *DefaultAccountManager) updateSingleGroup(ctx context.Context, accountI
}
events = am.prepareGroupEvents(ctx, transaction, accountID, userID, newGroup)
return nil
var err error
snap, err = affectedpeers.Load(ctx, transaction, accountID, change)
return err
})
return events, err
return events, snap, err
}
// prepareGroupEvents prepares a list of event functions to be stored.

2
management/server/networks/manager.go
View File

@@ -129,7 +129,7 @@ func (m *managerImpl) DeleteNetwork(ctx context.Context, accountID, userID, netw
var eventsToStore []func()
var snap *affectedpeers.Snapshot
change := affectedpeers.Change{NetworkIDs: []string{networkID}}
change := affectedpeers.Change{Networks: []*types.Network{network}}
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
// Load before the cascade deletes: pre-state still references the network.
var lerr error

31
management/server/networks/resources/manager.go
View File

@@ -116,7 +116,7 @@ func (m *managerImpl) CreateResource(ctx context.Context, userID string, resourc
var eventsToStore []func()
var snap *affectedpeers.Snapshot
change := affectedpeers.Change{ResourceIDs: []string{resource.ID}}
change := affectedpeers.Change{Resources: []*types.NetworkResource{resource}}
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
var txErr error
eventsToStore, snap, txErr = m.createResourceInTransaction(ctx, transaction, userID, resource, change)
@@ -256,9 +256,8 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc
if err != nil {
return fmt.Errorf("failed to get old resource groups: %w", err)
}
var oldGroupIDs []string
for _, g := range oldGroups {
oldGroupIDs = append(oldGroupIDs, g.ID)
oldResource.GroupIDs = append(oldResource.GroupIDs, g.ID)
}
err = transaction.SaveNetworkResource(ctx, resource)
@@ -276,12 +275,9 @@ func (m *managerImpl) UpdateResource(ctx context.Context, userID string, resourc
m.accountManager.StoreEvent(ctx, userID, resource.ID, resource.AccountID, activity.NetworkResourceUpdated, resource.EventMeta(network))
})
// Pass both old and new resource group IDs so policies that targeted the
// resource via a now-detached group still refresh their source peers.
change = affectedpeers.Change{
ResourceIDs: []string{resource.ID},
ChangedGroupIDs: append(oldGroupIDs, resource.GroupIDs...),
}
// Carry both the old and new resource so policies that targeted the resource
// via a now-detached group still refresh their source peers.
change = affectedpeers.Change{Resources: []*types.NetworkResource{oldResource, resource}}
if snap, err = affectedpeers.Load(ctx, transaction, resource.AccountID, change); err != nil {
return err
}
@@ -377,9 +373,22 @@ func (m *managerImpl) DeleteResource(ctx context.Context, accountID, userID, net
var events []func()
var snap *affectedpeers.Snapshot
change := affectedpeers.Change{ResourceIDs: []string{resourceID}}
var change affectedpeers.Change
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
// Load before delete: pre-state snapshot still references the resource.
// Capture the resource and its groups before delete: the post-delete state
// no longer references it.
existing, err := transaction.GetNetworkResourceByID(ctx, store.LockingStrengthUpdate, accountID, resourceID)
if err != nil {
return fmt.Errorf("failed to get network resource: %w", err)
}
oldGroups, err := m.groupsManager.GetResourceGroupsInTransaction(ctx, transaction, store.LockingStrengthNone, accountID, resourceID)
if err != nil {
return fmt.Errorf("failed to get resource groups: %w", err)
}
for _, g := range oldGroups {
existing.GroupIDs = append(existing.GroupIDs, g.ID)
}
change = affectedpeers.Change{Resources: []*types.NetworkResource{existing}}
if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil {
return err
}

68
management/server/networks/routers/manager.go
View File

@@ -92,7 +92,7 @@ func (m *managerImpl) CreateRouter(ctx context.Context, userID string, router *t
var network *networkTypes.Network
var snap *affectedpeers.Snapshot
change := affectedpeers.Change{NetworkIDs: []string{router.NetworkID}}
change := affectedpeers.Change{Routers: []*types.NetworkRouter{router}}
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
network, err = transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID)
if err != nil {
@@ -171,10 +171,9 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t
var network *networkTypes.Network
var snap *affectedpeers.Snapshot
var change affectedpeers.Change
var oldRouting []string
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
var txErr error
network, snap, change, oldRouting, txErr = m.updateRouterInTransaction(ctx, transaction, router)
network, snap, change, txErr = m.updateRouterInTransaction(ctx, transaction, router)
return txErr
})
if err != nil {
@@ -183,9 +182,7 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t
m.accountManager.StoreEvent(ctx, userID, router.ID, router.AccountID, activity.NetworkRouterUpdated, router.EventMeta(network))
// The previous routing peers lose their routing role and are no longer reachable
// from the post-update state, so union them as directly affected after Expand.
affectedPeerIDs := append(snap.Expand(ctx, router.AccountID, change), oldRouting...)
affectedPeerIDs := snap.Expand(ctx, router.AccountID, change)
if len(affectedPeerIDs) > 0 {
log.WithContext(ctx).Debugf("UpdateRouter %s: updating %d affected peers: %v", router.ID, len(affectedPeerIDs), affectedPeerIDs)
go m.accountManager.UpdateAffectedPeers(ctx, router.AccountID, affectedPeerIDs)
@@ -196,66 +193,43 @@ func (m *managerImpl) UpdateRouter(ctx context.Context, userID string, router *t
return router, nil
}
func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, *affectedpeers.Snapshot, affectedpeers.Change, []string, error) {
func (m *managerImpl) updateRouterInTransaction(ctx context.Context, transaction store.Store, router *types.NetworkRouter) (*networkTypes.Network, *affectedpeers.Snapshot, affectedpeers.Change, error) {
network, err := transaction.GetNetworkByID(ctx, store.LockingStrengthNone, router.AccountID, router.NetworkID)
if err != nil {
return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network: %w", err)
return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to get network: %w", err)
}
existing, err := transaction.GetNetworkRouterByID(ctx, store.LockingStrengthUpdate, router.AccountID, router.ID)
if err != nil {
return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to get network router: %w", err)
return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to get network router: %w", err)
}
if existing.AccountID != router.AccountID {
return nil, nil, affectedpeers.Change{}, nil, status.NewNetworkRouterNotFoundError(router.ID)
return nil, nil, affectedpeers.Change{}, status.NewNetworkRouterNotFoundError(router.ID)
}
if existing.NetworkID != router.NetworkID {
return nil, nil, affectedpeers.Change{}, nil, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID)
return nil, nil, affectedpeers.Change{}, status.NewRouterNotPartOfNetworkError(router.ID, router.NetworkID)
}
// Capture the previous routing peers before persisting the update.
oldRouting := oldRoutingPeerIDs(ctx, transaction, router.AccountID, existing)
if err = transaction.UpdateNetworkRouter(ctx, router); err != nil {
return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to update network router: %w", err)
return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to update network router: %w", err)
}
if err = transaction.IncrementNetworkSerial(ctx, router.AccountID); err != nil {
return nil, nil, affectedpeers.Change{}, nil, fmt.Errorf("failed to increment network serial: %w", err)
return nil, nil, affectedpeers.Change{}, fmt.Errorf("failed to increment network serial: %w", err)
}
networkIDs := []string{router.NetworkID}
if existing.NetworkID != router.NetworkID {
networkIDs = append(networkIDs, existing.NetworkID)
}
change := affectedpeers.Change{NetworkIDs: networkIDs}
// Carry both the previous and updated router so the bridge folds the old and
// new routing peers; a repoint loses the old peers' routing role and the
// post-update state can no longer reach them.
change := affectedpeers.Change{Routers: []*types.NetworkRouter{existing, router}}
snap, err := affectedpeers.Load(ctx, transaction, router.AccountID, change)
if err != nil {
return nil, nil, affectedpeers.Change{}, nil, err
return nil, nil, affectedpeers.Change{}, err
}
return network, snap, change, oldRouting, nil
}
// oldRoutingPeerIDs returns the peer IDs that served as the router's routing peers
// before an update (direct Peer plus PeerGroups members).
func oldRoutingPeerIDs(ctx context.Context, transaction store.Store, accountID string, existing *types.NetworkRouter) []string {
var ids []string
if existing.Peer != "" {
ids = append(ids, existing.Peer)
}
if len(existing.PeerGroups) > 0 {
groupPeers, err := transaction.GetPeerIDsByGroups(ctx, accountID, existing.PeerGroups)
if err != nil {
log.WithContext(ctx).Errorf("failed to get old router peer-group members for affected peers: %v", err)
} else {
ids = append(ids, groupPeers...)
}
}
return ids
return network, snap, change, nil
}
func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, networkID, routerID string) error {
@@ -269,9 +243,15 @@ func (m *managerImpl) DeleteRouter(ctx context.Context, accountID, userID, netwo
var event func()
var snap *affectedpeers.Snapshot
change := affectedpeers.Change{NetworkIDs: []string{networkID}}
var change affectedpeers.Change
err = m.store.ExecuteInTransaction(ctx, func(transaction store.Store) error {
// Load before delete: pre-state still references the router and its peers.
// Capture the router before delete: its peers lose their routing role and
// the post-delete state can no longer reach them.
existing, err := transaction.GetNetworkRouterByID(ctx, store.LockingStrengthUpdate, accountID, routerID)
if err != nil {
return fmt.Errorf("failed to get network router: %w", err)
}
change = affectedpeers.Change{Routers: []*types.NetworkRouter{existing}}
if snap, err = affectedpeers.Load(ctx, transaction, accountID, change); err != nil {
return err
}

27
management/server/peer.go
View File

@@ -1478,6 +1478,33 @@ func (am *DefaultAccountManager) expandAndUpdateAffected(ctx context.Context, ac
}
}
// dispatchAffected expands a set of (snapshot, change) pairs collected across
// several transactions, unions their affected peers, and dispatches one update.
// Used by batch operations that commit per-item transactions but want a single
// affected-peers pass — each snapshot is still loaded inside its own transaction.
func (am *DefaultAccountManager) dispatchAffected(ctx context.Context, accountID string, snaps []*affectedpeers.Snapshot, changes []affectedpeers.Change) {
seen := make(map[string]struct{})
var affectedPeerIDs []string
for i, snap := range snaps {
if snap == nil {
continue
}
for _, id := range snap.Expand(ctx, accountID, changes[i]) {
if _, ok := seen[id]; !ok {
seen[id] = struct{}{}
affectedPeerIDs = append(affectedPeerIDs, id)
}
}
}
if len(affectedPeerIDs) > 0 {
log.WithContext(ctx).Debugf("dispatchAffected: account %s updating %d affected peers: %v", accountID, len(affectedPeerIDs), affectedPeerIDs)
am.UpdateAffectedPeers(ctx, accountID, affectedPeerIDs)
} else {
log.WithContext(ctx).Tracef("dispatchAffected: account %s no affected peers", accountID)
}
}
// resolvePeerIDs resolves group IDs and direct peer IDs into a deduplicated peer ID list.
func (am *DefaultAccountManager) resolvePeerIDs(ctx context.Context, s store.Store, accountID string, groupIDs []string, directPeerIDs []string) []string {
peerIDs, err := s.GetPeerIDsByGroups(ctx, accountID, groupIDs)