[client] Feature/lazy connection (#3379)

With the lazy connection feature, the peer will connect to target peers on-demand. The trigger can be any IP traffic.

This feature can be enabled with the NB_ENABLE_EXPERIMENTAL_LAZY_CONN environment variable.

When the engine receives a network map, it binds a free UDP port for every remote peer, and the system configures WireGuard endpoints for these ports. When traffic appears on a UDP socket, the system removes this listener and starts the peer connection procedure immediately.

Key changes
Fix slow netbird status -d command
Move from engine.go file to conn_mgr.go the peer connection related code
Refactor the iface interface usage and moved interface file next to the engine code
Add new command line flag and UI option to enable feature
The peer.Conn struct is reusable after it has been closed.
Change connection states
Connection states
Idle: The peer is not attempting to establish a connection. This typically means it's in a lazy state or the remote peer is expired.

Connecting: The peer is actively trying to establish a connection. This occurs when the peer has entered an active state and is continuously attempting to reach the remote peer.

Connected: A successful peer-to-peer connection has been established and communication is active.

client/internal/lazyconn/activity/listen_ip.go

@@ -0,0 +1,9 @@
+//go:build !linux || android
+
+package activity
+
+import "net"
+
+var (
+	listenIP = net.IP{127, 0, 0, 1}
+)

client/internal/lazyconn/activity/listen_ip_linux.go

@@ -0,0 +1,10 @@
+//go:build !android
+
+package activity
+
+import "net"
+
+var (
+	// use this ip to avoid eBPF proxy congestion
+	listenIP = net.IP{127, 0, 1, 1}
+)

client/internal/lazyconn/activity/listener.go

@@ -0,0 +1,106 @@
+package activity
+
+import (
+	"fmt"
+	"net"
+	"sync"
+	"sync/atomic"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+)
+
+// Listener it is not a thread safe implementation, do not call Close before ReadPackets. It will cause blocking
+type Listener struct {
+	wgIface  lazyconn.WGIface
+	peerCfg  lazyconn.PeerConfig
+	conn     *net.UDPConn
+	endpoint *net.UDPAddr
+	done     sync.Mutex
+
+	isClosed atomic.Bool // use to avoid error log when closing the listener
+}
+
+func NewListener(wgIface lazyconn.WGIface, cfg lazyconn.PeerConfig) (*Listener, error) {
+	d := &Listener{
+		wgIface: wgIface,
+		peerCfg: cfg,
+	}
+
+	conn, err := d.newConn()
+	if err != nil {
+		return nil, fmt.Errorf("failed to creating activity listener: %v", err)
+	}
+	d.conn = conn
+	d.endpoint = conn.LocalAddr().(*net.UDPAddr)
+
+	if err := d.createEndpoint(); err != nil {
+		return nil, err
+	}
+	d.done.Lock()
+	cfg.Log.Infof("created activity listener: %s", conn.LocalAddr().(*net.UDPAddr).String())
+	return d, nil
+}
+
+func (d *Listener) ReadPackets() {
+	for {
+		n, remoteAddr, err := d.conn.ReadFromUDP(make([]byte, 1))
+		if err != nil {
+			if d.isClosed.Load() {
+				d.peerCfg.Log.Debugf("exit from activity listener")
+			} else {
+				d.peerCfg.Log.Errorf("failed to read from activity listener: %s", err)
+			}
+			break
+		}
+
+		if n < 1 {
+			d.peerCfg.Log.Warnf("received %d bytes from %s, too short", n, remoteAddr)
+			continue
+		}
+		break
+	}
+
+	if err := d.removeEndpoint(); err != nil {
+		d.peerCfg.Log.Errorf("failed to remove endpoint: %s", err)
+	}
+
+	_ = d.conn.Close() // do not care err because some cases it will return "use of closed network connection"
+	d.done.Unlock()
+}
+
+func (d *Listener) Close() {
+	d.peerCfg.Log.Infof("closing listener: %s", d.conn.LocalAddr().String())
+	d.isClosed.Store(true)
+
+	if err := d.conn.Close(); err != nil {
+		d.peerCfg.Log.Errorf("failed to close UDP listener: %s", err)
+	}
+	d.done.Lock()
+}
+
+func (d *Listener) removeEndpoint() error {
+	d.peerCfg.Log.Debugf("removing lazy endpoint: %s", d.endpoint.String())
+	return d.wgIface.RemovePeer(d.peerCfg.PublicKey)
+}
+
+func (d *Listener) createEndpoint() error {
+	d.peerCfg.Log.Debugf("creating lazy endpoint: %s", d.endpoint.String())
+	return d.wgIface.UpdatePeer(d.peerCfg.PublicKey, d.peerCfg.AllowedIPs, 0, d.endpoint, nil)
+}
+
+func (d *Listener) newConn() (*net.UDPConn, error) {
+	addr := &net.UDPAddr{
+		Port: 0,
+		IP:   listenIP,
+	}
+
+	conn, err := net.ListenUDP("udp", addr)
+	if err != nil {
+		log.Errorf("failed to create activity listener on %s: %s", addr, err)
+		return nil, err
+	}
+
+	return conn, nil
+}

client/internal/lazyconn/activity/listener_test.go

@@ -0,0 +1,41 @@
+package activity
+
+import (
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+)
+
+func TestNewListener(t *testing.T) {
+	peer := &MocPeer{
+		PeerID: "examplePublicKey1",
+	}
+
+	cfg := lazyconn.PeerConfig{
+		PublicKey:  peer.PeerID,
+		PeerConnID: peer.ConnID(),
+		Log:        log.WithField("peer", "examplePublicKey1"),
+	}
+
+	l, err := NewListener(MocWGIface{}, cfg)
+	if err != nil {
+		t.Fatalf("failed to create listener: %v", err)
+	}
+
+	chanClosed := make(chan struct{})
+	go func() {
+		defer close(chanClosed)
+		l.ReadPackets()
+	}()
+
+	time.Sleep(1 * time.Second)
+	l.Close()
+
+	select {
+	case <-chanClosed:
+	case <-time.After(time.Second):
+	}
+}

client/internal/lazyconn/activity/manager.go

@@ -0,0 +1,95 @@
+package activity
+
+import (
+	"sync"
+
+	log "github.com/sirupsen/logrus"
+
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+	peerid "github.com/netbirdio/netbird/client/internal/peer/id"
+)
+
+type Manager struct {
+	OnActivityChan chan peerid.ConnID
+
+	wgIface lazyconn.WGIface
+
+	peers map[peerid.ConnID]*Listener
+	done  chan struct{}
+
+	mu sync.Mutex
+}
+
+func NewManager(wgIface lazyconn.WGIface) *Manager {
+	m := &Manager{
+		OnActivityChan: make(chan peerid.ConnID, 1),
+		wgIface:        wgIface,
+		peers:          make(map[peerid.ConnID]*Listener),
+		done:           make(chan struct{}),
+	}
+	return m
+}
+
+func (m *Manager) MonitorPeerActivity(peerCfg lazyconn.PeerConfig) error {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	if _, ok := m.peers[peerCfg.PeerConnID]; ok {
+		log.Warnf("activity listener already exists for: %s", peerCfg.PublicKey)
+		return nil
+	}
+
+	listener, err := NewListener(m.wgIface, peerCfg)
+	if err != nil {
+		return err
+	}
+	m.peers[peerCfg.PeerConnID] = listener
+
+	go m.waitForTraffic(listener, peerCfg.PeerConnID)
+	return nil
+}
+
+func (m *Manager) RemovePeer(log *log.Entry, peerConnID peerid.ConnID) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	listener, ok := m.peers[peerConnID]
+	if !ok {
+		return
+	}
+	log.Debugf("removing activity listener")
+	delete(m.peers, peerConnID)
+	listener.Close()
+}
+
+func (m *Manager) Close() {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+
+	close(m.done)
+	for peerID, listener := range m.peers {
+		delete(m.peers, peerID)
+		listener.Close()
+	}
+}
+
+func (m *Manager) waitForTraffic(listener *Listener, peerConnID peerid.ConnID) {
+	listener.ReadPackets()
+
+	m.mu.Lock()
+	if _, ok := m.peers[peerConnID]; !ok {
+		m.mu.Unlock()
+		return
+	}
+	delete(m.peers, peerConnID)
+	m.mu.Unlock()
+
+	m.notify(peerConnID)
+}
+
+func (m *Manager) notify(peerConnID peerid.ConnID) {
+	select {
+	case <-m.done:
+	case m.OnActivityChan <- peerConnID:
+	}
+}

client/internal/lazyconn/activity/manager_test.go

@@ -0,0 +1,162 @@
+package activity
+
+import (
+	"net"
+	"net/netip"
+	"testing"
+	"time"
+
+	log "github.com/sirupsen/logrus"
+	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
+
+	"github.com/netbirdio/netbird/client/internal/lazyconn"
+	peerid "github.com/netbirdio/netbird/client/internal/peer/id"
+)
+
+type MocPeer struct {
+	PeerID string
+}
+
+func (m *MocPeer) ConnID() peerid.ConnID {
+	return peerid.ConnID(m)
+}
+
+type MocWGIface struct {
+}
+
+func (m MocWGIface) RemovePeer(string) error {
+	return nil
+}
+
+func (m MocWGIface) UpdatePeer(string, []netip.Prefix, time.Duration, *net.UDPAddr, *wgtypes.Key) error {
+	return nil
+
+}
+
+func TestManager_MonitorPeerActivity(t *testing.T) {
+	mocWgInterface := &MocWGIface{}
+
+	peer1 := &MocPeer{
+		PeerID: "examplePublicKey1",
+	}
+	mgr := NewManager(mocWgInterface)
+	defer mgr.Close()
+	peerCfg1 := lazyconn.PeerConfig{
+		PublicKey:  peer1.PeerID,
+		PeerConnID: peer1.ConnID(),
+		Log:        log.WithField("peer", "examplePublicKey1"),
+	}
+
+	if err := mgr.MonitorPeerActivity(peerCfg1); err != nil {
+		t.Fatalf("failed to monitor peer activity: %v", err)
+	}
+
+	if err := trigger(mgr.peers[peerCfg1.PeerConnID].conn.LocalAddr().String()); err != nil {
+		t.Fatalf("failed to trigger activity: %v", err)
+	}
+
+	select {
+	case peerConnID := <-mgr.OnActivityChan:
+		if peerConnID != peerCfg1.PeerConnID {
+			t.Fatalf("unexpected peerConnID: %v", peerConnID)
+		}
+	case <-time.After(1 * time.Second):
+	}
+}
+
+func TestManager_RemovePeerActivity(t *testing.T) {
+	mocWgInterface := &MocWGIface{}
+
+	peer1 := &MocPeer{
+		PeerID: "examplePublicKey1",
+	}
+	mgr := NewManager(mocWgInterface)
+	defer mgr.Close()
+
+	peerCfg1 := lazyconn.PeerConfig{
+		PublicKey:  peer1.PeerID,
+		PeerConnID: peer1.ConnID(),
+		Log:        log.WithField("peer", "examplePublicKey1"),
+	}
+
+	if err := mgr.MonitorPeerActivity(peerCfg1); err != nil {
+		t.Fatalf("failed to monitor peer activity: %v", err)
+	}
+
+	addr := mgr.peers[peerCfg1.PeerConnID].conn.LocalAddr().String()
+
+	mgr.RemovePeer(peerCfg1.Log, peerCfg1.PeerConnID)
+
+	if err := trigger(addr); err != nil {
+		t.Fatalf("failed to trigger activity: %v", err)
+	}
+
+	select {
+	case <-mgr.OnActivityChan:
+		t.Fatal("should not have active activity")
+	case <-time.After(1 * time.Second):
+	}
+}
+
+func TestManager_MultiPeerActivity(t *testing.T) {
+	mocWgInterface := &MocWGIface{}
+
+	peer1 := &MocPeer{
+		PeerID: "examplePublicKey1",
+	}
+	mgr := NewManager(mocWgInterface)
+	defer mgr.Close()
+
+	peerCfg1 := lazyconn.PeerConfig{
+		PublicKey:  peer1.PeerID,
+		PeerConnID: peer1.ConnID(),
+		Log:        log.WithField("peer", "examplePublicKey1"),
+	}
+
+	peer2 := &MocPeer{}
+	peerCfg2 := lazyconn.PeerConfig{
+		PublicKey:  peer2.PeerID,
+		PeerConnID: peer2.ConnID(),
+		Log:        log.WithField("peer", "examplePublicKey2"),
+	}
+
+	if err := mgr.MonitorPeerActivity(peerCfg1); err != nil {
+		t.Fatalf("failed to monitor peer activity: %v", err)
+	}
+
+	if err := mgr.MonitorPeerActivity(peerCfg2); err != nil {
+		t.Fatalf("failed to monitor peer activity: %v", err)
+	}
+
+	if err := trigger(mgr.peers[peerCfg1.PeerConnID].conn.LocalAddr().String()); err != nil {
+		t.Fatalf("failed to trigger activity: %v", err)
+	}
+
+	if err := trigger(mgr.peers[peerCfg2.PeerConnID].conn.LocalAddr().String()); err != nil {
+		t.Fatalf("failed to trigger activity: %v", err)
+	}
+
+	for i := 0; i < 2; i++ {
+		select {
+		case <-mgr.OnActivityChan:
+		case <-time.After(1 * time.Second):
+			t.Fatal("timed out waiting for activity")
+		}
+	}
+}
+
+func trigger(addr string) error {
+	// Create a connection to the destination UDP address and port
+	conn, err := net.Dial("udp", addr)
+	if err != nil {
+		return err
+	}
+	defer conn.Close()
+
+	// Write the bytes to the UDP connection
+	_, err = conn.Write([]byte{0x01, 0x02, 0x03, 0x04, 0x05})
+	if err != nil {
+		return err
+	}
+	return nil
+}