mirror of
https://github.com/netbirdio/netbird.git
synced 2026-10-23 22:09:09 +02:00
[client] Apply the local peer config and peer list notification once per map
Every bounded pass re-ran updateConfig, which notifies the address listener and logs at info level when SSH is disabled, and FinishPeerListModifications, which snapshots every peer for the router subscribers and wakes the UI. Both describe the map, not a batch: the config now applies on the first pass and the notification goes out when the map converges, or for what was applied when a pass fails.
This commit is contained in:
@@ -1690,9 +1690,8 @@ func (e *Engine) updateTURNs(turns []*mgmProto.ProtectedHostConfig) error {
|
|||||||
// than the cap, so the caller re-runs until convergence.
|
// than the cap, so the caller re-runs until convergence.
|
||||||
func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int, firstPass bool) (bool, error) {
|
func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int, firstPass bool) (bool, error) {
|
||||||
// intentionally leave it before checking serial because for now it can happen that peer IP changed but serial didn't
|
// intentionally leave it before checking serial because for now it can happen that peer IP changed but serial didn't
|
||||||
if networkMap.GetPeerConfig() != nil {
|
if firstPass && networkMap.GetPeerConfig() != nil {
|
||||||
err := e.updateConfig(networkMap.GetPeerConfig())
|
if err := e.updateConfig(networkMap.GetPeerConfig()); err != nil {
|
||||||
if err != nil {
|
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -1713,9 +1712,9 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int,
|
|||||||
doneOffline := e.phase("offline_peers")
|
doneOffline := e.phase("offline_peers")
|
||||||
e.updateOfflinePeers(networkMap.GetOfflinePeers())
|
e.updateOfflinePeers(networkMap.GetOfflinePeers())
|
||||||
doneOffline()
|
doneOffline()
|
||||||
}
|
|
||||||
|
|
||||||
log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers()))
|
log.Debugf("got peers update from Management Service, total peers to connect to = %d", len(networkMap.GetRemotePeers()))
|
||||||
|
}
|
||||||
|
|
||||||
// Filter out own peer from the remote peers list
|
// Filter out own peer from the remote peers list
|
||||||
localPubKey := e.config.WgPrivateKey.PublicKey().String()
|
localPubKey := e.config.WgPrivateKey.PublicKey().String()
|
||||||
@@ -1731,12 +1730,15 @@ func (e *Engine) updateNetworkMap(networkMap *mgmProto.NetworkMap, maxBatch int,
|
|||||||
// every peer in batches — same path as a normal update, no unbounded removeAllPeers
|
// every peer in batches — same path as a normal update, no unbounded removeAllPeers
|
||||||
// held under syncMsgMux in one shot.
|
// held under syncMsgMux in one shot.
|
||||||
needMore, err := e.applyPeerPhases(remotePeers, maxBatch)
|
needMore, err := e.applyPeerPhases(remotePeers, maxBatch)
|
||||||
|
if err != nil || !needMore {
|
||||||
|
// The peer list notification snapshots every peer and wakes the UI, so it
|
||||||
|
// goes out once per map, or for whatever was applied before a failure.
|
||||||
|
e.statusRecorder.FinishPeerListModifications()
|
||||||
|
}
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return false, err
|
return false, err
|
||||||
}
|
}
|
||||||
|
|
||||||
e.statusRecorder.FinishPeerListModifications()
|
|
||||||
|
|
||||||
// The SSH and lazy-exclude steps walk the whole map and need every target peer
|
// The SSH and lazy-exclude steps walk the whole map and need every target peer
|
||||||
// in the store, so they run once, on the pass that added the last batch. The
|
// in the store, so they run once, on the pass that added the last batch. The
|
||||||
// exclude list also has replace-semantics: a partial set would be wrong.
|
// exclude list also has replace-semantics: a partial set would be wrong.
|
||||||
|
|||||||
@@ -9,6 +9,7 @@ import (
|
|||||||
"os"
|
"os"
|
||||||
"strings"
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -828,8 +829,53 @@ func TestEngine_UpdateNetworkMap(t *testing.T) {
|
|||||||
assert.Equal(t, p.GetSshConfig().GetSshPubKey(), hostKey(p), "the converging pass should record the host key of %s", p.GetWgPubKey())
|
assert.Equal(t, p.GetSshConfig().GetSshPubKey(), hostKey(p), "the converging pass should record the host key of %s", p.GetWgPubKey())
|
||||||
}
|
}
|
||||||
})
|
})
|
||||||
|
|
||||||
|
newPeer := func(key, ip string) *mgmtProto.RemotePeerConfig {
|
||||||
|
return &mgmtProto.RemotePeerConfig{WgPubKey: key, AllowedIps: []string{ip}}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Local peer config handling notifies the UI listener. It belongs to the
|
||||||
|
// map, so a map drained over several passes notifies once.
|
||||||
|
t.Run("chunked apply handles the local peer config once", func(t *testing.T) {
|
||||||
|
listener := &countingListener{}
|
||||||
|
engine.statusRecorder.SetConnectionListener(listener)
|
||||||
|
defer engine.statusRecorder.RemoveConnectionListener()
|
||||||
|
// Registering replays the last known address to the new listener.
|
||||||
|
registered := listener.addressChanges.Load()
|
||||||
|
|
||||||
|
nm := &mgmtProto.NetworkMap{
|
||||||
|
Serial: 10,
|
||||||
|
PeerConfig: &mgmtProto.PeerConfig{Address: "10.20.0.1/24", Fqdn: "local.netbird.cloud"},
|
||||||
|
RemotePeers: []*mgmtProto.RemotePeerConfig{
|
||||||
|
newPeer("AAHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=", "100.64.0.30/32"),
|
||||||
|
newPeer("ABHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=", "100.64.0.31/32"),
|
||||||
|
newPeer("ACHf3Ma6z6mdLbriAJbqhX7+nM/B71lgw2+91q3LfhU=", "100.64.0.32/32"),
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
more, err := engine.updateNetworkMap(nm, 1, true)
|
||||||
|
require.NoError(t, err)
|
||||||
|
for i := 0; more && i < 20; i++ {
|
||||||
|
more, err = engine.updateNetworkMap(nm, 1, false)
|
||||||
|
require.NoError(t, err)
|
||||||
|
}
|
||||||
|
require.False(t, more, "the map should converge")
|
||||||
|
assert.EqualValues(t, 1, listener.addressChanges.Load()-registered, "the local peer config should be applied once per map")
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
type countingListener struct {
|
||||||
|
addressChanges atomic.Int32
|
||||||
|
}
|
||||||
|
|
||||||
|
func (l *countingListener) OnStateChanged(peer.ClientState) {}
|
||||||
|
func (l *countingListener) OnConnected() {}
|
||||||
|
func (l *countingListener) OnDisconnected() {}
|
||||||
|
func (l *countingListener) OnConnecting() {}
|
||||||
|
func (l *countingListener) OnDisconnecting() {}
|
||||||
|
func (l *countingListener) OnAddressChanged(string, string) { l.addressChanges.Add(1) }
|
||||||
|
func (l *countingListener) OnPeersListChanged(int) {}
|
||||||
|
|
||||||
func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) {
|
func TestEngine_UpdateNetworkMapWithRoutes(t *testing.T) {
|
||||||
testCases := []struct {
|
testCases := []struct {
|
||||||
name string
|
name string
|
||||||
|
|||||||
Reference in New Issue
Block a user