[management] added event store metrics

[client] Add TCP support to DNS forwarder service listener

client/internal/dnsfwd/forwarder.go

@@ -33,6 +33,8 @@ type DNSForwarder struct {
 
 	dnsServer *dns.Server
 	mux       *dns.ServeMux
+	tcpServer *dns.Server
+	tcpMux    *dns.ServeMux
 
 	mutex      sync.RWMutex
 	fwdEntries []*ForwarderEntry
@@ -50,22 +52,41 @@ func NewDNSForwarder(listenAddress string, ttl uint32, firewall firewall.Manager
 }
 
 func (f *DNSForwarder) Listen(entries []*ForwarderEntry) error {
-	log.Infof("listen DNS forwarder on address=%s", f.listenAddress)
-	mux := dns.NewServeMux()
+	log.Infof("starting DNS forwarder on address=%s", f.listenAddress)
 
-	dnsServer := &dns.Server{
+	// UDP server
+	mux := dns.NewServeMux()
+	f.mux = mux
+	f.dnsServer = &dns.Server{
 		Addr:    f.listenAddress,
 		Net:     "udp",
 		Handler: mux,
 	}
-	f.dnsServer = dnsServer
-	f.mux = mux
+	// TCP server
+	tcpMux := dns.NewServeMux()
+	f.tcpMux = tcpMux
+	f.tcpServer = &dns.Server{
+		Addr:    f.listenAddress,
+		Net:     "tcp",
+		Handler: tcpMux,
+	}
 
 	f.UpdateDomains(entries)
 
-	return dnsServer.ListenAndServe()
-}
+	errCh := make(chan error, 2)
 
+	go func() {
+		log.Infof("DNS UDP listener running on %s", f.listenAddress)
+		errCh <- f.dnsServer.ListenAndServe()
+	}()
+	go func() {
+		log.Infof("DNS TCP listener running on %s", f.listenAddress)
+		errCh <- f.tcpServer.ListenAndServe()
+	}()
+
+	// return the first error we get (e.g. bind failure or shutdown)
+	return <-errCh
+}
 func (f *DNSForwarder) UpdateDomains(entries []*ForwarderEntry) {
 	f.mutex.Lock()
 	defer f.mutex.Unlock()
@@ -77,31 +98,41 @@ func (f *DNSForwarder) UpdateDomains(entries []*ForwarderEntry) {
 	}
 
 	oldDomains := filterDomains(f.fwdEntries)
-
 	for _, d := range oldDomains {
 		f.mux.HandleRemove(d.PunycodeString())
+		f.tcpMux.HandleRemove(d.PunycodeString())
 	}
 
 	newDomains := filterDomains(entries)
 	for _, d := range newDomains {
-		f.mux.HandleFunc(d.PunycodeString(), f.handleDNSQuery)
+		f.mux.HandleFunc(d.PunycodeString(), f.handleDNSQueryUDP)
+		f.tcpMux.HandleFunc(d.PunycodeString(), f.handleDNSQueryTCP)
 	}
 
 	f.fwdEntries = entries
-
 	log.Debugf("Updated domains from %v to %v", oldDomains, newDomains)
 }
 
 func (f *DNSForwarder) Close(ctx context.Context) error {
-	if f.dnsServer == nil {
-		return nil
+	var result *multierror.Error
+
+	if f.dnsServer != nil {
+		if err := f.dnsServer.ShutdownContext(ctx); err != nil {
+			result = multierror.Append(result, fmt.Errorf("UDP shutdown: %w", err))
+		}
 	}
-	return f.dnsServer.ShutdownContext(ctx)
+	if f.tcpServer != nil {
+		if err := f.tcpServer.ShutdownContext(ctx); err != nil {
+			result = multierror.Append(result, fmt.Errorf("TCP shutdown: %w", err))
+		}
+	}
+
+	return nberrors.FormatErrorOrNil(result)
 }
 
-func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) {
+func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) *dns.Msg {
 	if len(query.Question) == 0 {
-		return
+		return nil
 	}
 	question := query.Question[0]
 	log.Tracef("received DNS request for DNS forwarder: domain=%v type=%v class=%v",
@@ -123,20 +154,53 @@ func (f *DNSForwarder) handleDNSQuery(w dns.ResponseWriter, query *dns.Msg) {
 		if err := w.WriteMsg(resp); err != nil {
 			log.Errorf("failed to write DNS response: %v", err)
 		}
-		return
+		return nil
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), upstreamTimeout)
 	defer cancel()
 	ips, err := net.DefaultResolver.LookupNetIP(ctx, network, domain)
 	if err != nil {
-		f.handleDNSError(w, resp, domain, err)
-		return
+		f.handleDNSError(w, query, resp, domain, err)
+		return nil
 	}
 
 	f.updateInternalState(domain, ips)
 	f.addIPsToResponse(resp, domain, ips)
 
+	return resp
+}
+
+func (f *DNSForwarder) handleDNSQueryUDP(w dns.ResponseWriter, query *dns.Msg) {
+
+	resp := f.handleDNSQuery(w, query)
+	if resp == nil {
+		return
+	}
+
+	opt := query.IsEdns0()
+	maxSize := dns.MinMsgSize
+	if opt != nil {
+		// client advertised a larger EDNS0 buffer
+		maxSize = int(opt.UDPSize())
+	}
+
+	// if our response is too big, truncate and set the TC bit
+	if resp.Len() > maxSize {
+		resp.Truncate(maxSize)
+	}
+
+	if err := w.WriteMsg(resp); err != nil {
+		log.Errorf("failed to write DNS response: %v", err)
+	}
+}
+
+func (f *DNSForwarder) handleDNSQueryTCP(w dns.ResponseWriter, query *dns.Msg) {
+	resp := f.handleDNSQuery(w, query)
+	if resp == nil {
+		return
+	}
+
 	if err := w.WriteMsg(resp); err != nil {
 		log.Errorf("failed to write DNS response: %v", err)
 	}
@@ -179,7 +243,7 @@ func (f *DNSForwarder) updateFirewall(matchingEntries []*ForwarderEntry, prefixe
 }
 
 // handleDNSError processes DNS lookup errors and sends an appropriate error response
-func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, resp *dns.Msg, domain string, err error) {
+func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, query, resp *dns.Msg, domain string, err error) {
 	var dnsErr *net.DNSError
 
 	switch {
@@ -191,7 +255,7 @@ func (f *DNSForwarder) handleDNSError(w dns.ResponseWriter, resp *dns.Msg, domai
 		}
 
 		if dnsErr.Server != "" {
-			log.Warnf("failed to resolve query for domain=%s server=%s: %v", domain, dnsErr.Server, err)
+			log.Warnf("failed to resolve query for type=%s domain=%s server=%s: %v", dns.TypeToString[query.Question[0].Qtype], domain, dnsErr.Server, err)
 		} else {
 			log.Warnf(errResolveFailed, domain, err)
 		}

client/internal/dnsfwd/manager.go

@@ -33,6 +33,7 @@ type Manager struct {
 	statusRecorder *peer.Status
 
 	fwRules      []firewall.Rule
+	tcpRules     []firewall.Rule
 	dnsForwarder *DNSForwarder
 }
 
@@ -107,6 +108,13 @@ func (m *Manager) allowDNSFirewall() error {
 	}
 	m.fwRules = dnsRules
 
+	tcpRules, err := m.firewall.AddPeerFiltering(nil, net.IP{0, 0, 0, 0}, firewall.ProtocolTCP, nil, dport, firewall.ActionAccept, "")
+	if err != nil {
+		log.Errorf("failed to add allow DNS router rules, err: %v", err)
+		return err
+	}
+	m.tcpRules = tcpRules
+
 	return nil
 }
 
@@ -117,7 +125,13 @@ func (m *Manager) dropDNSFirewall() error {
 			mErr = multierror.Append(mErr, fmt.Errorf("failed to delete DNS router rules, err: %v", err))
 		}
 	}
+	for _, rule := range m.tcpRules {
+		if err := m.firewall.DeletePeerRule(rule); err != nil {
+			mErr = multierror.Append(mErr, fmt.Errorf("failed to delete DNS router rules, err: %v", err))
+		}
+	}
 
 	m.fwRules = nil
+	m.tcpRules = nil
 	return nberrors.FormatErrorOrNil(mErr)
 }

client/internal/networkmonitor/check_change_bsd.go

@@ -19,7 +19,7 @@ import (
 func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) error {
 	fd, err := unix.Socket(syscall.AF_ROUTE, syscall.SOCK_RAW, syscall.AF_UNSPEC)
 	if err != nil {
-		return fmt.Errorf("failed to open routing socket: %v", err)
+		return fmt.Errorf("open routing socket: %v", err)
 	}
 	defer func() {
 		err := unix.Close(fd)

client/internal/networkmonitor/check_change_windows.go

@@ -13,7 +13,7 @@ import (
 func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) error {
 	routeMonitor, err := systemops.NewRouteMonitor(ctx)
 	if err != nil {
-		return fmt.Errorf("failed to create route monitor: %w", err)
+		return fmt.Errorf("create route monitor: %w", err)
 	}
 	defer func() {
 		if err := routeMonitor.Stop(); err != nil {
@@ -38,35 +38,49 @@ func checkChange(ctx context.Context, nexthopv4, nexthopv6 systemops.Nexthop) er
 }
 
 func routeChanged(route systemops.RouteUpdate, nexthopv4, nexthopv6 systemops.Nexthop) bool {
-	intf := "<nil>"
-	if route.Interface != nil {
-		intf = route.Interface.Name
-		if isSoftInterface(intf) {
-			log.Debugf("Network monitor: ignoring default route change for soft interface %s", intf)
-			return false
-		}
+	if intf := route.NextHop.Intf; intf != nil && isSoftInterface(intf.Name) {
+		log.Debugf("Network monitor: ignoring default route change for next hop with soft interface %s", route.NextHop)
+		return false
+	}
+
+	// TODO: for the empty nexthop ip (on-link), determine the family differently
+	nexthop := nexthopv4
+	if route.NextHop.IP.Is6() {
+		nexthop = nexthopv6
 	}
 
 	switch route.Type {
-	case systemops.RouteModified:
-		// TODO: get routing table to figure out if our route is affected for modified routes
-		log.Infof("Network monitor: default route changed: via %s, interface %s", route.NextHop, intf)
-		return true
-	case systemops.RouteAdded:
-		if route.NextHop.Is4() && route.NextHop != nexthopv4.IP || route.NextHop.Is6() && route.NextHop != nexthopv6.IP {
-			log.Infof("Network monitor: default route added: via %s, interface %s", route.NextHop, intf)
-			return true
-		}
+	case systemops.RouteModified, systemops.RouteAdded:
+		return handleRouteAddedOrModified(route, nexthop)
 	case systemops.RouteDeleted:
-		if nexthopv4.Intf != nil && route.NextHop == nexthopv4.IP || nexthopv6.Intf != nil && route.NextHop == nexthopv6.IP {
-			log.Infof("Network monitor: default route removed: via %s, interface %s", route.NextHop, intf)
-			return true
-		}
+		return handleRouteDeleted(route, nexthop)
 	}
 
 	return false
 }
 
+func handleRouteAddedOrModified(route systemops.RouteUpdate, nexthop systemops.Nexthop) bool {
+	// For added/modified routes, we care about different next hops
+	if !nexthop.Equal(route.NextHop) {
+		action := "changed"
+		if route.Type == systemops.RouteAdded {
+			action = "added"
+		}
+		log.Infof("Network monitor: default route %s: via %s", action, route.NextHop)
+		return true
+	}
+	return false
+}
+
+func handleRouteDeleted(route systemops.RouteUpdate, nexthop systemops.Nexthop) bool {
+	// For deleted routes, we care about our tracked next hop being deleted
+	if nexthop.Equal(route.NextHop) {
+		log.Infof("Network monitor: default route removed: via %s", route.NextHop)
+		return true
+	}
+	return false
+}
+
 func isSoftInterface(name string) bool {
 	return strings.Contains(strings.ToLower(name), "isatap") || strings.Contains(strings.ToLower(name), "teredo")
 }

client/internal/networkmonitor/check_change_windows_test.go

@@ -0,0 +1,404 @@
+package networkmonitor
+
+import (
+	"net"
+	"net/netip"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+
+	"github.com/netbirdio/netbird/client/internal/routemanager/systemops"
+)
+
+func TestRouteChanged(t *testing.T) {
+	tests := []struct {
+		name      string
+		route     systemops.RouteUpdate
+		nexthopv4 systemops.Nexthop
+		nexthopv6 systemops.Nexthop
+		expected  bool
+	}{
+		{
+			name: "soft interface should be ignored",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{
+						Name: "ISATAP-Interface", // isSoftInterface checks name
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.2"),
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+			},
+			expected: false,
+		},
+		{
+			name: "modified route with different v4 nexthop IP should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.2"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+			},
+			expected: true,
+		},
+		{
+			name: "modified route with same v4 nexthop (IP and Intf Index) should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+			},
+			expected: false,
+		},
+		{
+			name: "added route with different v6 nexthop IP should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteAdded,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("2001:db8::2"),
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.1"),
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			expected: true,
+		},
+		{
+			name: "added route with same v6 nexthop (IP and Intf Index) should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteAdded,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("2001:db8::1"),
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.1"),
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			expected: false,
+		},
+		{
+			name: "deleted route matching tracked v4 nexthop (IP and Intf Index) should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+			},
+			expected: true,
+		},
+		{
+			name: "deleted route not matching tracked v4 nexthop (different IP) should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP: netip.MustParseAddr("192.168.1.3"), // Different IP
+					Intf: &net.Interface{
+						Index: 1, Name: "eth0",
+					},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP: netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{
+					Index: 1, Name: "eth0",
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP: netip.MustParseAddr("2001:db8::1"),
+			},
+			expected: false,
+		},
+		{
+			name: "modified v4 route with same IP, different Intf Index should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP:   netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: true,
+		},
+		{
+			name: "modified v4 route with same IP, one Intf nil, other non-nil should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("192.168.1.1"),
+					Intf: nil, // Intf is nil
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP:   netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"}, // Tracked Intf is not nil
+			},
+			expected: true,
+		},
+		{
+			name: "added v4 route with same IP, different Intf Index should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteAdded,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP:   netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: true,
+		},
+		{
+			name: "deleted v4 route with same IP, different Intf Index should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{ // This is the route being deleted
+					IP:   netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv4: systemops.Nexthop{ // This is our tracked nexthop
+				IP:   netip.MustParseAddr("192.168.1.1"),
+				Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
+			},
+			expected: false, // Because nexthopv4.Equal(route.NextHop) will be false
+		},
+		{
+			name: "modified v6 route with different IP, same Intf Index should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("2001:db8::3"), // Different IP
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: true,
+		},
+		{
+			name: "modified v6 route with same IP, different Intf Index should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("2001:db8::1"),
+					Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: true,
+		},
+		{
+			name: "modified v6 route with same IP, same Intf Index should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteModified,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("2001:db8::1"),
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: false,
+		},
+		{
+			name: "deleted v6 route matching tracked nexthop (IP and Intf Index) should return true",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("2001:db8::1"),
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: true,
+		},
+		{
+			name: "deleted v6 route not matching tracked nexthop (different IP) should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("2001:db8::3"), // Different IP
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv6: systemops.Nexthop{
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: false,
+		},
+		{
+			name: "deleted v6 route not matching tracked nexthop (same IP, different Intf Index) should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteDeleted,
+				Destination: netip.PrefixFrom(netip.IPv6Unspecified(), 0),
+				NextHop: systemops.Nexthop{ // This is the route being deleted
+					IP:   netip.MustParseAddr("2001:db8::1"),
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv6: systemops.Nexthop{ // This is our tracked nexthop
+				IP:   netip.MustParseAddr("2001:db8::1"),
+				Intf: &net.Interface{Index: 2, Name: "eth1"}, // Different Intf Index
+			},
+			expected: false,
+		},
+		{
+			name: "unknown route type should return false",
+			route: systemops.RouteUpdate{
+				Type:        systemops.RouteUpdateType(99), // Unknown type
+				Destination: netip.PrefixFrom(netip.IPv4Unspecified(), 0),
+				NextHop: systemops.Nexthop{
+					IP:   netip.MustParseAddr("192.168.1.1"),
+					Intf: &net.Interface{Index: 1, Name: "eth0"},
+				},
+			},
+			nexthopv4: systemops.Nexthop{
+				IP:   netip.MustParseAddr("192.168.1.2"), // Different from route.NextHop
+				Intf: &net.Interface{Index: 1, Name: "eth0"},
+			},
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := routeChanged(tt.route, tt.nexthopv4, tt.nexthopv6)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}
+
+func TestIsSoftInterface(t *testing.T) {
+	tests := []struct {
+		name     string
+		ifname   string
+		expected bool
+	}{
+		{
+			name:     "ISATAP interface should be detected",
+			ifname:   "ISATAP tunnel adapter",
+			expected: true,
+		},
+		{
+			name:     "lowercase soft interface should be detected",
+			ifname:   "isatap.{14A5CF17-CA72-43EC-B4EA-B4B093641B7D}",
+			expected: true,
+		},
+		{
+			name:     "Teredo interface should be detected",
+			ifname:   "Teredo Tunneling Pseudo-Interface",
+			expected: true,
+		},
+		{
+			name:     "regular interface should not be detected as soft",
+			ifname:   "eth0",
+			expected: false,
+		},
+		{
+			name:     "another regular interface should not be detected as soft",
+			ifname:   "wlan0",
+			expected: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			result := isSoftInterface(tt.ifname)
+			assert.Equal(t, tt.expected, result)
+		})
+	}
+}

client/internal/networkmonitor/monitor.go

@@ -118,9 +118,12 @@ func (nw *NetworkMonitor) Stop() {
 }
 
 func (nw *NetworkMonitor) checkChanges(ctx context.Context, event chan struct{}, nexthop4 systemops.Nexthop, nexthop6 systemops.Nexthop) {
+	defer close(event)
 	for {
 		if err := checkChangeFn(ctx, nexthop4, nexthop6); err != nil {
-			close(event)
+			if !errors.Is(err, context.Canceled) {
+				log.Errorf("Network monitor: failed to check for changes: %v", err)
+			}
 			return
 		}
 		// prevent blocking

client/internal/routemanager/systemops/systemops.go

@@ -1,6 +1,7 @@
 package systemops
 
 import (
+	"fmt"
 	"net"
 	"net/netip"
 	"sync"
@@ -15,6 +16,20 @@ type Nexthop struct {
 	Intf *net.Interface
 }
 
+// Equal checks if two nexthops are equal.
+func (n Nexthop) Equal(other Nexthop) bool {
+	return n.IP == other.IP && (n.Intf == nil && other.Intf == nil ||
+		n.Intf != nil && other.Intf != nil && n.Intf.Index == other.Intf.Index)
+}
+
+// String returns a string representation of the nexthop.
+func (n Nexthop) String() string {
+	if n.Intf == nil {
+		return n.IP.String()
+	}
+	return fmt.Sprintf("%s @ %d (%s)", n.IP.String(), n.Intf.Index, n.Intf.Name)
+}
+
 type ExclusionCounter = refcounter.Counter[netip.Prefix, struct{}, Nexthop]
 
 type SysOps struct {

client/internal/routemanager/systemops/systemops_windows.go

@@ -33,8 +33,7 @@ type RouteUpdateType int
 type RouteUpdate struct {
 	Type        RouteUpdateType
 	Destination netip.Prefix
-	NextHop     netip.Addr
-	Interface   *net.Interface
+	NextHop     Nexthop
 }
 
 // RouteMonitor provides a way to monitor changes in the routing table.
@@ -231,15 +230,15 @@ func (rm *RouteMonitor) parseUpdate(row *MIB_IPFORWARD_ROW2, notificationType MI
 		intf, err := net.InterfaceByIndex(idx)
 		if err != nil {
 			log.Warnf("failed to get interface name for index %d: %v", idx, err)
-			update.Interface = &net.Interface{
+			update.NextHop.Intf = &net.Interface{
 				Index: idx,
 			}
 		} else {
-			update.Interface = intf
+			update.NextHop.Intf = intf
 		}
 	}
 
-	log.Tracef("Received route update with destination %v, next hop %v, interface %v", row.DestinationPrefix, row.NextHop, update.Interface)
+	log.Tracef("Received route update with destination %v, next hop %v, interface %v", row.DestinationPrefix, row.NextHop, update.NextHop.Intf)
 	dest := parseIPPrefix(row.DestinationPrefix, idx)
 	if !dest.Addr().IsValid() {
 		return RouteUpdate{}, fmt.Errorf("invalid destination: %v", row)
@@ -262,7 +261,7 @@ func (rm *RouteMonitor) parseUpdate(row *MIB_IPFORWARD_ROW2, notificationType MI
 
 	update.Type = updateType
 	update.Destination = dest
-	update.NextHop = nexthop
+	update.NextHop.IP = nexthop
 
 	return update, nil
 }

go.mod

@@ -106,7 +106,6 @@ require (
 	golang.org/x/oauth2 v0.24.0
 	golang.org/x/sync v0.13.0
 	golang.org/x/term v0.31.0
-	golang.org/x/time v0.5.0
 	google.golang.org/api v0.177.0
 	gopkg.in/yaml.v3 v3.0.1
 	gorm.io/driver/mysql v1.5.7
@@ -241,6 +240,7 @@ require (
 	golang.org/x/image v0.18.0 // indirect
 	golang.org/x/mod v0.17.0 // indirect
 	golang.org/x/text v0.24.0 // indirect
+	golang.org/x/time v0.5.0 // indirect
 	golang.org/x/tools v0.21.1-0.20240508182429-e35e4ccd0d2d // indirect
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2 // indirect
 	google.golang.org/genproto/googleapis/api v0.0.0-20240509183442-62759503f434 // indirect

management/cmd/management.go

@@ -134,7 +134,6 @@ var (
 		},
 		RunE: func(cmd *cobra.Command, args []string) error {
 			flag.Parse()
-			startPprof()
 
 			ctx, cancel := context.WithCancel(cmd.Context())
 			defer cancel()
@@ -177,7 +176,7 @@ var (
 			if disableSingleAccMode {
 				mgmtSingleAccModeDomain = ""
 			}
-			eventStore, key, err := integrations.InitEventStore(ctx, config.Datadir, config.DataStoreEncryptionKey)
+			eventStore, key, err := integrations.InitEventStore(ctx, config.Datadir, config.DataStoreEncryptionKey, appMetrics)
 			if err != nil {
 				return fmt.Errorf("failed to initialize database: %s", err)
 			}

management/cmd/root.go

@@ -2,13 +2,9 @@ package cmd
 
 import (
 	"fmt"
-	"net/http"
-	_ "net/http/pprof"
 	"os"
 	"os/signal"
-	"runtime"
 
-	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
 	"github.com/netbirdio/netbird/management/server/types"
@@ -21,17 +17,6 @@ const (
 	idpSignKeyRefreshEnabledFlagName = "idp-sign-key-refresh-enabled"
 )
 
-func startPprof() {
-	go func() {
-		runtime.SetBlockProfileRate(1)
-		runtime.SetMutexProfileFraction(1)
-		log.Debugf("Starting pprof server on 0.0.0.0:6060")
-		if err := http.ListenAndServe("0.0.0.0:6060", nil); err != nil {
-			log.Fatalf("pprof server failed: %v", err)
-		}
-	}()
-}
-
 var (
 	dnsDomain                string
 	mgmtDataDir              string

management/server/account.go

@@ -202,7 +202,7 @@ func BuildManager(
 	if err != nil {
 		initialInterval = 1
 	} else {
-		initialInterval = int64(interval) * 2
+		initialInterval = int64(interval) * 10
 		go func() {
 			time.Sleep(30 * time.Second)
 			am.updateAccountPeersBufferInterval.Store(int64(time.Duration(interval) * time.Millisecond))

management/server/account/manager.go

@@ -117,5 +117,4 @@ type Manager interface {
 	UpdateToPrimaryAccount(ctx context.Context, accountId string) (*types.Account, error)
 	GetOwnerInfo(ctx context.Context, accountId string) (*types.UserInfo, error)
 	GetCurrentUserInfo(ctx context.Context, userAuth nbcontext.UserAuth) (*users.UserInfoWithPermissions, error)
-	BufferUpdateAccountPeers(ctx context.Context, accountID string)
 }

management/server/grpcserver.go

@@ -3,11 +3,8 @@ package server
 import (
 	"context"
 	"fmt"
-	"math/rand/v2"
 	"net"
 	"net/netip"
-	"os"
-	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -16,7 +13,6 @@ import (
 	"github.com/golang/protobuf/ptypes/timestamp"
 	"github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/realip"
 	log "github.com/sirupsen/logrus"
-	"golang.org/x/time/rate"
 	"golang.zx2c4.com/wireguard/wgctrl/wgtypes"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/peer"
@@ -51,11 +47,6 @@ type GRPCServer struct {
 	ephemeralManager   *EphemeralManager
 	peerLocks          sync.Map
 	authManager        auth.Manager
-
-	syncLimiter       *rate.Limiter
-	loginLimiter      *rate.Limiter
-	loginLimiterStore sync.Map
-	loginPeerLimit    rate.Limit
 }
 
 // NewServer creates a new Management server
@@ -85,41 +76,6 @@ func NewServer(
 		}
 	}
 
-	multiplier := time.Second
-	d, e := time.ParseDuration(os.Getenv("NB_LOGIN_RATE"))
-	if e == nil {
-		multiplier = d
-	}
-
-	loginRatePerS, err := strconv.Atoi(os.Getenv("NB_LOGIN_RATE_PER_M"))
-	if loginRatePerS == 0 || err != nil {
-		loginRatePerS = 200
-	}
-
-	loginBurst, err := strconv.Atoi(os.Getenv("NB_LOGIN_BURST"))
-	if loginBurst == 0 || err != nil {
-		loginBurst = 200
-	}
-	log.WithContext(ctx).Infof("login burst limit set to %d", loginBurst)
-
-	loginPeerRatePerS, err := strconv.Atoi(os.Getenv("NB_LOGIN_PEER_RATE_PER_M"))
-	if loginPeerRatePerS == 0 || err != nil {
-		loginPeerRatePerS = 200
-	}
-	log.WithContext(ctx).Infof("login rate limit set to %d/min", loginRatePerS)
-
-	syncRatePerS, err := strconv.Atoi(os.Getenv("NB_SYNC_RATE_PER_M"))
-	if syncRatePerS == 0 || err != nil {
-		syncRatePerS = 200
-	}
-	log.WithContext(ctx).Infof("sync rate limit set to %d/min", syncRatePerS)
-
-	syncBurst, err := strconv.Atoi(os.Getenv("NB_SYNC_BURST"))
-	if syncBurst == 0 || err != nil {
-		syncBurst = 200
-	}
-	log.WithContext(ctx).Infof("sync burst limit set to %d", syncBurst)
-
 	return &GRPCServer{
 		wgKey: key,
 		// peerKey -> event channel
@@ -131,9 +87,6 @@ func NewServer(
 		authManager:        authManager,
 		appMetrics:         appMetrics,
 		ephemeralManager:   ephemeralManager,
-		syncLimiter:        rate.NewLimiter(rate.Every(time.Minute/time.Duration(syncRatePerS)), syncBurst),
-		loginLimiter:       rate.NewLimiter(rate.Every(multiplier/time.Duration(loginRatePerS)), loginBurst),
-		loginPeerLimit:     rate.Every(time.Minute / time.Duration(loginPeerRatePerS)),
 	}, nil
 }
 
@@ -175,18 +128,11 @@ func getRealIP(ctx context.Context) net.IP {
 // Sync validates the existence of a connecting peer, sends an initial state (all available for the connecting peers) and
 // notifies the connected peer of any updates (e.g. new peers under the same account)
 func (s *GRPCServer) Sync(req *proto.EncryptedMessage, srv proto.ManagementService_SyncServer) error {
+	reqStart := time.Now()
 	if s.appMetrics != nil {
 		s.appMetrics.GRPCMetrics().CountSyncRequest()
 	}
 
-	if !s.syncLimiter.Allow() {
-		time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
-		log.Warnf("sync rate limit exceeded for peer %s", req.WgPubKey)
-		return status.Errorf(codes.Internal, "temp rate limit reached")
-	}
-
-	reqStart := time.Now()
-
 	ctx := srv.Context()
 
 	syncReq := &proto.SyncRequest{}
@@ -470,58 +416,15 @@ func (s *GRPCServer) parseRequest(ctx context.Context, req *proto.EncryptedMessa
 // In case it isn't, the endpoint checks whether setup key is provided within the request and tries to register a peer.
 // In case of the successful registration login is also successful
 func (s *GRPCServer) Login(ctx context.Context, req *proto.EncryptedMessage) (*proto.EncryptedMessage, error) {
-	if s.appMetrics != nil {
-		s.appMetrics.GRPCMetrics().CountLoginRequest()
-	}
-
-	limiterIface, ok := s.loginLimiterStore.Load(req.WgPubKey)
-	if !ok {
-		// Check global limiter before allowing a new peer limiter
-		if !s.loginLimiter.Allow() {
-			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
-			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
-			return nil, fmt.Errorf("temp rate limit reached (global limit)")
-		}
-
-		// Create new limiter for this peer
-		newLimiter := rate.NewLimiter(s.loginPeerLimit, 1000)
-		s.loginLimiterStore.Store(req.WgPubKey, newLimiter)
-
-		if !newLimiter.Allow() {
-			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
-			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
-			return nil, fmt.Errorf("temp rate limit reached (new peer limit)")
-		}
-	} else {
-		// Use existing limiter for this peer
-		limiter := limiterIface.(*rate.Limiter)
-		if !limiter.Allow() {
-			time.Sleep(time.Second + (time.Millisecond * time.Duration(rand.IntN(20)*100)))
-			log.WithContext(ctx).Warnf("rate limit exceeded for peer %s", req.WgPubKey)
-			return nil, fmt.Errorf("temp rate limit reached (peer limit)")
-		}
-	}
-
-	// limiter, _ := s.loginLimiterStore.LoadOrStore(req.WgPubKey, rate.NewLimiter(s.loginPeerLimit, 1))
-	// if !limiter.(*rate.Limiter).Allow() {
-	//	time.Sleep(time.Millisecond * time.Duration(rand.IntN(10)*100))
-	//	log.WithContext(ctx).Warnf("rate limit exceeded for %s", req.WgPubKey)
-	//	return nil, status.Errorf(codes.Internal, "temp rate limit reached")
-	// }
-	//
-	// if os.Getenv("ENABLE_LOGIN_RATE_LIMIT") == "true" {
-	//	if !s.loginLimiter.Allow() {
-	//		return nil, status.Errorf(codes.Internal, "temp rate limit reached")
-	//	}
-	// }
-
 	reqStart := time.Now()
 	defer func() {
 		if s.appMetrics != nil {
 			s.appMetrics.GRPCMetrics().CountLoginRequestDuration(time.Since(reqStart))
 		}
 	}()
-
+	if s.appMetrics != nil {
+		s.appMetrics.GRPCMetrics().CountLoginRequest()
+	}
 	realIP := getRealIP(ctx)
 	log.WithContext(ctx).Debugf("Login request from peer [%s] [%s]", req.WgPubKey, realIP.String())
 

management/server/http/handlers/peers/peers_handler.go

@@ -4,17 +4,10 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"net"
 	"net/http"
-	"os"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
 
 	"github.com/gorilla/mux"
 	log "github.com/sirupsen/logrus"
-	"golang.org/x/time/rate"
 
 	"github.com/netbirdio/netbird/management/server/account"
 	"github.com/netbirdio/netbird/management/server/activity"
@@ -30,9 +23,6 @@ import (
 // Handler is a handler that returns peers of the account
 type Handler struct {
 	accountManager account.Manager
-	rateLimiter    *rate.Limiter
-	limiterStore   sync.Map
-	reqLimit       rate.Limit
 }
 
 func AddEndpoints(accountManager account.Manager, router *mux.Router) {
@@ -45,15 +35,8 @@ func AddEndpoints(accountManager account.Manager, router *mux.Router) {
 
 // NewHandler creates a new peers Handler
 func NewHandler(accountManager account.Manager) *Handler {
-	apiRatePerM, err := strconv.Atoi(os.Getenv("NB_API_RATE_PER_M"))
-	if apiRatePerM == 0 || err != nil {
-		apiRatePerM = 60
-	}
-	log.Infof("peers API rate limit set to %d/min", apiRatePerM)
 	return &Handler{
 		accountManager: accountManager,
-		rateLimiter:    rate.NewLimiter(rate.Every(time.Minute/time.Duration(apiRatePerM)), 1),
-		reqLimit:       rate.Every(time.Minute / time.Duration(apiRatePerM)),
 	}
 }
 
@@ -71,11 +54,6 @@ func (h *Handler) checkPeerStatus(peer *nbpeer.Peer) (*nbpeer.Peer, error) {
 }
 
 func (h *Handler) getPeer(ctx context.Context, accountID, peerID, userID string, w http.ResponseWriter) {
-	if !h.rateLimiter.Allow() {
-		util.WriteError(ctx, fmt.Errorf("temp rate limit reached"), w)
-		return
-	}
-
 	peer, err := h.accountManager.GetPeer(ctx, accountID, peerID, userID)
 	if err != nil {
 		util.WriteError(ctx, err, w)
@@ -113,7 +91,7 @@ func (h *Handler) updatePeer(ctx context.Context, accountID, userID, peerID stri
 	req := &api.PeerRequest{}
 	err := json.NewDecoder(r.Body).Decode(&req)
 	if err != nil {
-		util.WriteErrorResponse("couldn't parse JSON request", http.StatusPreconditionRequired, w)
+		util.WriteErrorResponse("couldn't parse JSON request", http.StatusBadRequest, w)
 		return
 	}
 
@@ -206,40 +184,9 @@ func (h *Handler) HandlePeer(w http.ResponseWriter, r *http.Request) {
 		util.WriteError(r.Context(), status.Errorf(status.NotFound, "unknown METHOD"), w)
 	}
 }
-func getCallerIP(r *http.Request) string {
-	// Check X-Forwarded-For header first (can be a comma-separated list)
-	if xff := r.Header.Get("X-Forwarded-For"); xff != "" {
-		// Use first IP in the list
-		parts := strings.Split(xff, ",")
-		return strings.TrimSpace(parts[0])
-	}
-
-	// Then check X-Real-IP
-	if xrip := r.Header.Get("X-Real-IP"); xrip != "" {
-		return xrip
-	}
-
-	// Fallback to RemoteAddr
-	ip, _, err := net.SplitHostPort(r.RemoteAddr)
-	if err != nil {
-		return r.RemoteAddr // may be raw IP
-	}
-	return ip
-}
 
 // GetAllPeers returns a list of all peers associated with a provided account
 func (h *Handler) GetAllPeers(w http.ResponseWriter, r *http.Request) {
-	ip := getCallerIP(r)
-	limiter, _ := h.limiterStore.LoadOrStore(ip, rate.NewLimiter(h.reqLimit, 1))
-	if !limiter.(*rate.Limiter).Allow() {
-		log.WithContext(r.Context()).Errorf("rate limit exceeded for IP: %s", ip)
-		util.WriteError(r.Context(), status.Errorf(status.StatusTooManyRequests, "temp rate limit reached"), w)
-		return
-	}
-	//if !h.rateLimiter.Allow() {
-	//	util.WriteError(r.Context(), status.Errorf(status.StatusTooManyRequests, "temp rate limit reached"), w)
-	//	return
-	//}
 	userAuth, err := nbcontext.GetUserAuthFromContext(r.Context())
 	if err != nil {
 		util.WriteError(r.Context(), err, w)

management/server/http/util/util.go

@@ -106,8 +106,6 @@ func WriteError(ctx context.Context, err error, w http.ResponseWriter) {
 			httpStatus = http.StatusUnauthorized
 		case status.BadRequest:
 			httpStatus = http.StatusBadRequest
-		case status.StatusTooManyRequests:
-			httpStatus = http.StatusTooManyRequests
 		default:
 		}
 		msg = strings.ToLower(err.Error())

management/server/integrations/port_forwarding/controller.go

@@ -3,14 +3,12 @@ package port_forwarding
 import (
 	"context"
 
-	"github.com/netbirdio/netbird/management/server/peer"
 	nbtypes "github.com/netbirdio/netbird/management/server/types"
 )
 
 type Controller interface {
-	SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string, accountPeers map[string]*peer.Peer)
-	GetProxyNetworkMaps(ctx context.Context, accountID, peerID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error)
-	GetProxyNetworkMapsAll(ctx context.Context, accountID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error)
+	SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string)
+	GetProxyNetworkMaps(ctx context.Context, accountID string) (map[string]*nbtypes.NetworkMap, error)
 	IsPeerInIngressPorts(ctx context.Context, accountID, peerID string) (bool, error)
 }
 
@@ -21,15 +19,11 @@ func NewControllerMock() *ControllerMock {
 	return &ControllerMock{}
 }
 
-func (c *ControllerMock) SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string, accountPeers map[string]*peer.Peer) {
+func (c *ControllerMock) SendUpdate(ctx context.Context, accountID string, affectedProxyID string, affectedPeerIDs []string) {
 	// noop
 }
 
-func (c *ControllerMock) GetProxyNetworkMaps(ctx context.Context, accountID, peerID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error) {
-	return make(map[string]*nbtypes.NetworkMap), nil
-}
-
-func (c *ControllerMock) GetProxyNetworkMapsAll(ctx context.Context, accountID string, accountPeers map[string]*peer.Peer) (map[string]*nbtypes.NetworkMap, error) {
+func (c *ControllerMock) GetProxyNetworkMaps(ctx context.Context, accountID string) (map[string]*nbtypes.NetworkMap, error) {
 	return make(map[string]*nbtypes.NetworkMap), nil
 }
 

management/server/mock_server/account_mock.go

@@ -889,7 +889,3 @@ func (am *MockAccountManager) GetCurrentUserInfo(ctx context.Context, userAuth n
 	}
 	return nil, status.Errorf(codes.Unimplemented, "method GetCurrentUserInfo is not implemented")
 }
-
-func (am *MockAccountManager) BufferUpdateAccountPeers(ctx context.Context, accountID string) {
-	// noop
-}

management/server/peer.go

@@ -419,7 +419,7 @@ func (am *DefaultAccountManager) GetNetworkMap(ctx context.Context, peerID strin
 	}
 	customZone := account.GetPeersCustomZone(ctx, am.GetDNSDomain(account.Settings))
 
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id, peerID, account.Peers)
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return nil, err
@@ -453,7 +453,6 @@ func (am *DefaultAccountManager) GetPeerNetwork(ctx context.Context, peerID stri
 // Each new Peer will be assigned a new next net.IP from the Account.Network and Account.Network.LastIP will be updated (IP's are not reused).
 // The peer property is just a placeholder for the Peer properties to pass further
 func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID string, peer *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, error) {
-	startGlobal := time.Now()
 	if setupKey == "" && userID == "" {
 		// no auth method provided => reject access
 		return nil, nil, nil, status.Errorf(status.Unauthenticated, "no peer auth method provided, please use a setup key or interactive SSO login")
@@ -506,8 +505,6 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 		var ephemeral bool
 		var groupsToAdd []string
 		var allowExtraDNSLabels bool
-
-		start := time.Now()
 		if addedByUser {
 			user, err := transaction.GetUserByUserID(ctx, store.LockingStrengthUpdate, userID)
 			if err != nil {
@@ -540,9 +537,6 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: setup key get took %v", time.Since(start))
-
-		start = time.Now()
 		if (strings.ToLower(peer.Meta.Hostname) == "iphone" || strings.ToLower(peer.Meta.Hostname) == "ipad") && userID != "" {
 			if am.idpManager != nil {
 				userdata, err := am.idpManager.GetUserDataByID(ctx, userID, idp.AppMetadata{WTAccountID: accountID})
@@ -551,21 +545,16 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 				}
 			}
 		}
-		log.WithContext(ctx).Debugf("AddPeer: idp took %v", time.Since(start))
 
-		start = time.Now()
 		freeLabel, err := am.getFreeDNSLabel(ctx, transaction, accountID, peer.Meta.Hostname)
 		if err != nil {
 			return fmt.Errorf("failed to get free DNS label: %w", err)
 		}
-		log.WithContext(ctx).Debugf("AddPeer: free label took %v", time.Since(start))
 
-		start = time.Now()
 		freeIP, err := getFreeIP(ctx, transaction, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to get free IP: %w", err)
 		}
-		log.WithContext(ctx).Debugf("AddPeer: ip took %v", time.Since(start))
 
 		registrationTime := time.Now().UTC()
 		newPeer = &nbpeer.Peer{
@@ -589,22 +578,17 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			ExtraDNSLabels:              peer.ExtraDNSLabels,
 			AllowExtraDNSLabels:         allowExtraDNSLabels,
 		}
-
-		start = time.Now()
 		settings, err := transaction.GetAccountSettings(ctx, store.LockingStrengthShare, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to get account settings: %w", err)
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: settings took %v", time.Since(start))
-
 		opEvent.TargetID = newPeer.ID
 		opEvent.Meta = newPeer.EventMeta(am.GetDNSDomain(settings))
 		if !addedByUser {
 			opEvent.Meta["setup_key_name"] = setupKeyName
 		}
 
-		start = time.Now()
 		if am.geo != nil && newPeer.Location.ConnectionIP != nil {
 			location, err := am.geo.Lookup(newPeer.Location.ConnectionIP)
 			if err != nil {
@@ -616,11 +600,8 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: geo took %v", time.Since(start))
-
 		newPeer = am.integratedPeerValidator.PreparePeer(ctx, accountID, newPeer, groupsToAdd, settings.Extra)
 
-		start = time.Now()
 		err = transaction.AddPeerToAllGroup(ctx, store.LockingStrengthUpdate, accountID, newPeer.ID)
 		if err != nil {
 			return fmt.Errorf("failed adding peer to All group: %w", err)
@@ -635,16 +616,11 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: add peer to group took %v", time.Since(start))
-
-		start = time.Now()
 		err = transaction.AddPeerToAccount(ctx, store.LockingStrengthUpdate, newPeer)
 		if err != nil {
 			return fmt.Errorf("failed to add peer to account: %w", err)
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: add peer to account took %v", time.Since(start))
-
 		err = transaction.IncrementNetworkSerial(ctx, store.LockingStrengthUpdate, accountID)
 		if err != nil {
 			return fmt.Errorf("failed to increment network serial: %w", err)
@@ -662,14 +638,11 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 			}
 		}
 
-		start = time.Now()
 		updateAccountPeers, err = isPeerInActiveGroup(ctx, transaction, accountID, newPeer.ID)
 		if err != nil {
 			return err
 		}
 
-		log.WithContext(ctx).Debugf("AddPeer: is peer in active group took %v", time.Since(start))
-
 		log.WithContext(ctx).Debugf("Peer %s added to account %s", newPeer.ID, accountID)
 		return nil
 	})
@@ -684,7 +657,8 @@ func (am *DefaultAccountManager) AddPeer(ctx context.Context, setupKey, userID s
 
 	am.StoreEvent(ctx, opEvent.InitiatorID, opEvent.TargetID, opEvent.AccountID, opEvent.Activity, opEvent.Meta)
 
-	log.WithContext(ctx).Debugf("AddPeer took %v", time.Since(startGlobal))
+	unlock()
+	unlock = nil
 
 	if updateAccountPeers {
 		am.BufferUpdateAccountPeers(ctx, accountID)
@@ -1023,56 +997,48 @@ func (am *DefaultAccountManager) checkIFPeerNeedsLoginWithoutLock(ctx context.Co
 }
 
 func (am *DefaultAccountManager) getValidatedPeerWithMap(ctx context.Context, isRequiresApproval bool, accountID string, peer *nbpeer.Peer) (*nbpeer.Peer, *types.NetworkMap, []*posture.Checks, error) {
-	mstart := time.Now()
+	start := time.Now()
 	defer func() {
-		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(mstart))
+		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(start))
 	}()
 
 	if isRequiresApproval {
-		start := time.Now()
 		network, err := am.Store.GetAccountNetwork(ctx, store.LockingStrengthShare, accountID)
 		if err != nil {
 			return nil, nil, nil, err
 		}
 
-		log.WithContext(ctx).Debugf("getValidatedPeerWithMap: took %s", time.Since(start))
-
 		emptyMap := &types.NetworkMap{
 			Network: network.Copy(),
 		}
 		return peer, emptyMap, nil, nil
 	}
-	start := time.Now()
+
 	account, err := am.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	log.WithContext(ctx).Debugf("GetAccountWithBackpressure: took %s", time.Since(start))
-	start = time.Now()
+
 	approvedPeersMap, err := am.integratedPeerValidator.GetValidatedPeers(account.Id, maps.Values(account.Groups), maps.Values(account.Peers), account.Settings.Extra)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	log.WithContext(ctx).Debugf("GetValidatedPeers: took %s", time.Since(start))
-	start = time.Now()
+
 	postureChecks, err := am.getPeerPostureChecks(account, peer.ID)
 	if err != nil {
 		return nil, nil, nil, err
 	}
-	log.WithContext(ctx).Debugf("getPeerPostureChecks: took %s", time.Since(start))
-	start = time.Now()
+
 	customZone := account.GetPeersCustomZone(ctx, am.GetDNSDomain(account.Settings))
-	log.WithContext(ctx).Debugf("GetPeersCustomZone: took %s", time.Since(start))
-	start = time.Now()
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id, peer.ID, account.Peers)
+
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, account.Id)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return nil, nil, nil, err
 	}
-	log.WithContext(ctx).Debugf("GetProxyNetworkMaps: took %s", time.Since(start))
-	start = time.Now()
+
 	networkMap := account.GetPeerNetworkMap(ctx, peer.ID, customZone, approvedPeersMap, account.GetResourcePoliciesMap(), account.GetResourceRoutersMap(), am.metrics.AccountManagerMetrics())
-	log.WithContext(ctx).Debugf("GetPeerNetworkMap: took %s", time.Since(start))
+
 	proxyNetworkMap, ok := proxyNetworkMaps[peer.ID]
 	if ok {
 		networkMap.Merge(proxyNetworkMap)
@@ -1200,16 +1166,13 @@ func (am *DefaultAccountManager) checkIfUserOwnsPeer(ctx context.Context, accoun
 // UpdateAccountPeers updates all peers that belong to an account.
 // Should be called when changes have to be synced to peers.
 func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, accountID string) {
-	globalStart := time.Now()
-	start := time.Now()
 	account, err := am.requestBuffer.GetAccountWithBackpressure(ctx, accountID)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to send out updates to peers. failed to get account: %v", err)
 		return
 	}
-	log.WithContext(ctx).Infof("updateAccountPeers: getAccount took %s", time.Since(start))
 
-	start = time.Now()
+	start := time.Now()
 
 	approvedPeersMap, err := am.integratedPeerValidator.GetValidatedPeers(account.Id, maps.Values(account.Groups), maps.Values(account.Peers), account.Settings.Extra)
 	if err != nil {
@@ -1217,8 +1180,6 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 		return
 	}
 
-	log.WithContext(ctx).Infof("updateAccountPeers: validatePeers took %s", time.Since(start))
-
 	var wg sync.WaitGroup
 	semaphore := make(chan struct{}, 10)
 
@@ -1228,21 +1189,11 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 	resourcePolicies := account.GetResourcePoliciesMap()
 	routers := account.GetResourceRoutersMap()
 
-	start = time.Now()
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMapsAll(ctx, accountID, account.Peers)
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountID)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return
 	}
-	log.WithContext(ctx).Infof("updateAccountPeers: getProxyNetworkMaps took %s", time.Since(start))
-	for _, id := range []string{"d07kd1ei389c73dq19gg", "d07kcaui389c73dq19g0", "d0e7uo6i389c73f040v0"} {
-		peerMap, ok := proxyNetworkMaps[id]
-		if !ok {
-			log.WithContext(ctx).Infof("updateAccountPeers xxx: proxy network map %s not found", id)
-			continue
-		}
-		log.WithContext(ctx).Infof("updateAccountPeers xxx: peer %s has %d peers, %d offline peers, %d, firewall rules, %d forwarding rules, %d routing rules", id, len(peerMap.Peers), len(peerMap.OfflinePeers), len(peerMap.FirewallRules), len(peerMap.ForwardingRules), len(peerMap.RoutesFirewallRules))
-	}
 
 	for _, peer := range account.Peers {
 		if !am.peersUpdateManager.HasChannel(peer.ID) {
@@ -1275,22 +1226,16 @@ func (am *DefaultAccountManager) UpdateAccountPeers(ctx context.Context, account
 				return
 			}
 
-			start = time.Now()
 			update := toSyncResponse(ctx, nil, p, nil, nil, remotePeerNetworkMap, dnsDomain, postureChecks, dnsCache, account.Settings.RoutingPeerDNSResolutionEnabled, extraSetting)
-			log.WithContext(ctx).Infof("updateAccountPeers: toSyncResponse took %s", time.Since(start))
-			start = time.Now()
 			am.peersUpdateManager.SendUpdate(ctx, p.ID, &UpdateMessage{Update: update, NetworkMap: remotePeerNetworkMap})
-			log.WithContext(ctx).Infof("updateAccountPeers: sending update toSyncResponse took %s", time.Since(start))
 		}(peer)
 	}
 
 	//
 
 	wg.Wait()
-	log.WithContext(ctx).Infof("updateAccountPeers: waiting for updates to complete took %s", time.Since(globalStart))
-
 	if am.metrics != nil {
-		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(globalStart))
+		am.metrics.AccountManagerMetrics().CountUpdateAccountPeersDuration(time.Since(start))
 	}
 }
 
@@ -1347,7 +1292,7 @@ func (am *DefaultAccountManager) UpdateAccountPeer(ctx context.Context, accountI
 		return
 	}
 
-	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountId, peerId, account.Peers)
+	proxyNetworkMaps, err := am.proxyController.GetProxyNetworkMaps(ctx, accountId)
 	if err != nil {
 		log.WithContext(ctx).Errorf("failed to get proxy network maps: %v", err)
 		return

management/server/peer/peer.go

@@ -24,7 +24,7 @@ type Peer struct {
 	// Meta is a Peer system meta data
 	Meta PeerSystemMeta `gorm:"embedded;embeddedPrefix:meta_"`
 	// Name is peer's name (machine name)
-	Name string `gorm:"index"`
+	Name string
 	// DNSLabel is the parsed peer name for domain resolution. It is used to form an FQDN by appending the account's
 	// domain to the peer label. e.g. peer-dns-label.netbird.cloud
 	DNSLabel string

management/server/status/error.go

@@ -37,8 +37,6 @@ const (
 
 	// Unauthenticated indicates that user is not authenticated due to absence of valid credentials
 	Unauthenticated Type = 10
-
-	StatusTooManyRequests = 11
 )
 
 // Type is a type of the Error

management/server/store/sql_store.go

@@ -1311,7 +1311,7 @@ func (s *SqlStore) GetAccountPeers(ctx context.Context, lockStrength LockingStre
 	query := s.db.Clauses(clause.Locking{Strength: string(lockStrength)}).Where(accountIDCondition, accountID)
 
 	if nameFilter != "" {
-		query = query.Where("name = ?", nameFilter)
+		query = query.Where("name LIKE ?", "%"+nameFilter+"%")
 	}
 	if ipFilter != "" {
 		query = query.Where("ip LIKE ?", "%"+ipFilter+"%")

management/server/telemetry/app_metrics.go

@@ -29,6 +29,7 @@ type MockAppMetrics struct {
 	StoreMetricsFunc             func() *StoreMetrics
 	UpdateChannelMetricsFunc     func() *UpdateChannelMetrics
 	AddAccountManagerMetricsFunc func() *AccountManagerMetrics
+	EventStreamingMetricsFunc    func() *EventStreamingMetrics
 }
 
 // GetMeter mocks the GetMeter function of the AppMetrics interface
@@ -103,6 +104,14 @@ func (mock *MockAppMetrics) AccountManagerMetrics() *AccountManagerMetrics {
 	return nil
 }
 
+// EventStreamingMetrics mocks the EventStreamingMetrics function of the AppMetrics interface
+func (mock *MockAppMetrics) EventStreamingMetrics() *EventStreamingMetrics {
+	if mock.EventStreamingMetricsFunc != nil {
+		return mock.EventStreamingMetricsFunc()
+	}
+	return nil
+}
+
 // AppMetrics is metrics interface
 type AppMetrics interface {
 	GetMeter() metric2.Meter
@@ -114,6 +123,7 @@ type AppMetrics interface {
 	StoreMetrics() *StoreMetrics
 	UpdateChannelMetrics() *UpdateChannelMetrics
 	AccountManagerMetrics() *AccountManagerMetrics
+	EventStreamingMetrics() *EventStreamingMetrics
 }
 
 // defaultAppMetrics are core application metrics based on OpenTelemetry https://opentelemetry.io/
@@ -128,6 +138,7 @@ type defaultAppMetrics struct {
 	storeMetrics          *StoreMetrics
 	updateChannelMetrics  *UpdateChannelMetrics
 	accountManagerMetrics *AccountManagerMetrics
+	eventStreamingMetrics *EventStreamingMetrics
 }
 
 // IDPMetrics returns metrics for the idp package
@@ -160,6 +171,16 @@ func (appMetrics *defaultAppMetrics) AccountManagerMetrics() *AccountManagerMetr
 	return appMetrics.accountManagerMetrics
 }
 
+// EventStreamingMetrics returns metrics for the event streaming module
+func (appMetrics *defaultAppMetrics) EventStreamingMetrics() *EventStreamingMetrics {
+	return appMetrics.eventStreamingMetrics
+}
+
+// RegisterEventStreamingMetrics sets the event streaming metrics
+func (appMetrics *defaultAppMetrics) RegisterEventStreamingMetrics(metrics *EventStreamingMetrics) {
+	appMetrics.eventStreamingMetrics = metrics
+}
+
 // Close stop application metrics HTTP handler and closes listener.
 func (appMetrics *defaultAppMetrics) Close() error {
 	if appMetrics.listener == nil {
@@ -184,10 +205,10 @@ func (appMetrics *defaultAppMetrics) Expose(ctx context.Context, port int, endpo
 	}
 	appMetrics.listener = listener
 	go func() {
-		err := http.Serve(listener, rootRouter)
-		if err != nil {
-			return
+		if err := http.Serve(listener, rootRouter); err != nil && err != http.ErrServerClosed {
+			log.WithContext(ctx).Errorf("metrics server error: %v", err)
 		}
+		log.WithContext(ctx).Info("metrics server stopped")
 	}()
 
 	log.WithContext(ctx).Infof("enabled application metrics and exposing on http://%s", listener.Addr().String())
@@ -204,7 +225,7 @@ func (appMetrics *defaultAppMetrics) GetMeter() metric2.Meter {
 func NewDefaultAppMetrics(ctx context.Context) (AppMetrics, error) {
 	exporter, err := prometheus.New()
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to create prometheus exporter: %w", err)
 	}
 
 	provider := metric.NewMeterProvider(metric.WithReader(exporter))
@@ -213,32 +234,32 @@ func NewDefaultAppMetrics(ctx context.Context) (AppMetrics, error) {
 
 	idpMetrics, err := NewIDPMetrics(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize IDP metrics: %w", err)
 	}
 
 	middleware, err := NewMetricsMiddleware(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize HTTP middleware metrics: %w", err)
 	}
 
 	grpcMetrics, err := NewGRPCMetrics(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize gRPC metrics: %w", err)
 	}
 
 	storeMetrics, err := NewStoreMetrics(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize store metrics: %w", err)
 	}
 
 	updateChannelMetrics, err := NewUpdateChannelMetrics(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize update channel metrics: %w", err)
 	}
 
 	accountManagerMetrics, err := NewAccountManagerMetrics(ctx, meter)
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to initialize account manager metrics: %w", err)
 	}
 
 	return &defaultAppMetrics{

sharedsock/sock_linux.go

@@ -234,7 +234,7 @@ func (s *SharedSocket) read(receiver receiver) {
 }
 
 // ReadFrom reads packets received in the packetDemux channel
-func (s *SharedSocket) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
+func (s *SharedSocket) ReadFrom(b []byte) (int, net.Addr, error) {
 	var pkt rcvdPacket
 	select {
 	case <-s.ctx.Done():
@@ -263,8 +263,7 @@ func (s *SharedSocket) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
 
 	decodedLayers := make([]gopacket.LayerType, 0, 3)
 
-	err = parser.DecodeLayers(pkt.buf, &decodedLayers)
-	if err != nil {
+	if err := parser.DecodeLayers(pkt.buf, &decodedLayers); err != nil {
 		return 0, nil, err
 	}
 
@@ -273,8 +272,8 @@ func (s *SharedSocket) ReadFrom(b []byte) (n int, addr net.Addr, err error) {
 		Port: int(udp.SrcPort),
 	}
 
-	copy(b, payload)
-	return int(udp.Length), remoteAddr, nil
+	n := copy(b, payload)
+	return n, remoteAddr, nil
 }
 
 // WriteTo builds a UDP packet and writes it using the specific IP version writer