sendnrw/netbird
2
0
Fork 0
mirror of https://github.com/netbirdio/netbird.git synced 2026-09-01 20:41:28 +02:00
Code Issues Packages Projects Releases Wiki Activity

merge main

Browse Source
This commit is contained in:
crn4
2026-05-29 14:08:09 +02:00
parent 7048b87931 174dc24867
commit eab0826b4e
29 changed files with 2675 additions and 1498 deletions
Download Patch File Download Diff File Expand all files Collapse all files

16
management/internals/controllers/network_map/controller/controller.go
View File

@@ -133,7 +133,7 @@ func (c *Controller) CountStreams() int {
return c.peersUpdateManager.CountStreams()
}
func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID string) error {
func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID string, reason types.UpdateReason) error {
log.WithContext(ctx).Tracef("updating peers for account %s from %s", accountID, util.GetCallerName())
account, err := c.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
if err != nil {
@@ -196,6 +196,10 @@ func (c *Controller) sendUpdateAccountPeers(ctx context.Context, accountID strin
continue
}
if c.accountManagerMetrics != nil {
c.accountManagerMetrics.CountNmapTriggered(string(reason.Resource), string(reason.Operation))
}
wg.Add(1)
semaphore <- struct{}{}
go func(p *nbpeer.Peer) {
@@ -271,14 +275,14 @@ func (c *Controller) bufferSendUpdateAccountPeers(ctx context.Context, accountID
go func() {
defer b.mu.Unlock()
_ = c.sendUpdateAccountPeers(ctx, accountID)
_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
if !b.update.Load() {
return
}
b.update.Store(false)
if b.next == nil {
b.next = time.AfterFunc(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
_ = c.sendUpdateAccountPeers(ctx, accountID)
_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
})
return
}
@@ -294,7 +298,7 @@ func (c *Controller) UpdateAccountPeers(ctx context.Context, accountID string, r
if c.accountManagerMetrics != nil {
c.accountManagerMetrics.CountUpdateAccountPeersTriggered(string(reason.Resource), string(reason.Operation))
}
return c.sendUpdateAccountPeers(ctx, accountID)
return c.sendUpdateAccountPeers(ctx, accountID, reason)
}
func (c *Controller) UpdateAccountPeer(ctx context.Context, accountId string, peerId string) error {
@@ -393,14 +397,14 @@ func (c *Controller) BufferUpdateAccountPeers(ctx context.Context, accountID str
go func() {
defer b.mu.Unlock()
_ = c.sendUpdateAccountPeers(ctx, accountID)
_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
if !b.update.Load() {
return
}
b.update.Store(false)
if b.next == nil {
b.next = time.AfterFunc(time.Duration(c.updateAccountPeersBufferInterval.Load()), func() {
_ = c.sendUpdateAccountPeers(ctx, accountID)
_ = c.sendUpdateAccountPeers(ctx, accountID, reason)
})
return
}