mirror of
https://github.com/netbirdio/netbird.git
synced 2026-07-08 15:49:58 +00:00
Compare commits
4 Commits
wg_watcher
...
v0.74.3
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7cd5c1732b | ||
|
|
816d80602f | ||
|
|
c9d387bd0d | ||
|
|
3aa6c02b93 |
@@ -85,7 +85,11 @@ func (g *Guard) reconnectLoopWithRetry(ctx context.Context, callback func()) {
|
||||
defer g.srWatcher.RemoveListener(srReconnectedChan)
|
||||
|
||||
ticker := g.initialTicker(ctx)
|
||||
defer ticker.Stop()
|
||||
defer func() {
|
||||
// If backoff.Ticker.send is blocked, context.Done will not close the Ticker goroutine.
|
||||
// We have to explicitly call Stop, even if we use backoff.WithContext.
|
||||
ticker.Stop()
|
||||
}()
|
||||
|
||||
tickerChannel := ticker.C
|
||||
|
||||
|
||||
92
client/internal/peer/guard/guard_leak_test.go
Normal file
92
client/internal/peer/guard/guard_leak_test.go
Normal file
@@ -0,0 +1,92 @@
|
||||
package guard
|
||||
|
||||
import (
|
||||
"context"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
|
||||
"github.com/netbirdio/netbird/client/internal/peer/ice"
|
||||
)
|
||||
|
||||
func newTestGuard(status connStatusFunc) *Guard {
|
||||
srw := NewSRWatcher(nil, nil, nil, ice.Config{})
|
||||
return NewGuard(log.WithField("test", "guard"), status, 50*time.Millisecond, srw)
|
||||
}
|
||||
|
||||
// countBackoffTickerGoroutines returns how many goroutines are currently sitting
|
||||
// in backoff/v4.(*Ticker).run (a ticker goroutine that has not exited).
|
||||
func countBackoffTickerGoroutines() int {
|
||||
buf := make([]byte, 1<<25) // 32MB
|
||||
n := runtime.Stack(buf, true)
|
||||
return strings.Count(string(buf[:n]), "backoff/v4.(*Ticker).run")
|
||||
}
|
||||
|
||||
// TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown reproduces a observed
|
||||
// leak: after a shutdown burst, ticker run/send goroutines stay parked
|
||||
// forever even though every reconnect loop has exited.
|
||||
func TestGuard_ReconnectTicker_NoGoroutineLeakOnShutdown(t *testing.T) {
|
||||
before := countBackoffTickerGoroutines()
|
||||
|
||||
const peers = 6000
|
||||
cancels := make([]context.CancelFunc, 0, peers)
|
||||
var wg sync.WaitGroup
|
||||
|
||||
// A status check slower than the tick cadence. This models the real
|
||||
// isConnectedOnAllWay/callback doing work: while the loop is busy in the
|
||||
// handler, the ticker fires the next tick and parks in send(), because
|
||||
// send() never selects on ctx.
|
||||
slowStatus := func() ConnStatus {
|
||||
time.Sleep(70 * time.Millisecond)
|
||||
return ConnStatusConnected
|
||||
}
|
||||
|
||||
for range peers {
|
||||
g := newTestGuard(slowStatus)
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
cancels = append(cancels, cancel)
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
g.Start(ctx, func() {})
|
||||
}()
|
||||
// Force the live ticker to be a newReconnectTicker.
|
||||
g.SetRelayedConnDisconnected()
|
||||
}
|
||||
|
||||
// Let the replacement tickers get past their 800ms initial interval, so
|
||||
// many are parked in send() waiting on the (slow) consumer when we tear
|
||||
// everything down.
|
||||
time.Sleep(1500 * time.Millisecond)
|
||||
|
||||
// Shutdown burst: cancel every peer at once, like engine teardown.
|
||||
for _, c := range cancels {
|
||||
c()
|
||||
}
|
||||
|
||||
// Every reconnect loop must return
|
||||
waitCh := make(chan struct{})
|
||||
go func() { wg.Wait(); close(waitCh) }()
|
||||
select {
|
||||
case <-waitCh:
|
||||
case <-time.After(30 * time.Second):
|
||||
t.Fatal("not all reconnect loops returned after ctx cancel")
|
||||
}
|
||||
|
||||
// Give any correctly-stopped ticker goroutines time to unwind.
|
||||
for range 50 {
|
||||
runtime.Gosched()
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
}
|
||||
|
||||
leaked := countBackoffTickerGoroutines() - before
|
||||
t.Logf("backoff Ticker.run goroutines still parked after teardown of %d peers: %d", peers, leaked)
|
||||
if leaked > 0 {
|
||||
t.Errorf("LEAK: %d backoff ticker goroutines parked after all reconnect loops exited "+
|
||||
"(defer ticker.Stop() stops the initial ticker, not the live replacement)", leaked)
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package server
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
log "github.com/sirupsen/logrus"
|
||||
@@ -181,6 +182,37 @@ func conflictBool(key string, p *bool) conflictCheck {
|
||||
}
|
||||
}
|
||||
|
||||
func canonicalURL(s string) string {
|
||||
u, err := url.ParseRequestURI(s)
|
||||
if err != nil {
|
||||
return s
|
||||
}
|
||||
if u.Port() == "" {
|
||||
switch u.Scheme {
|
||||
case "https":
|
||||
u.Host += ":443"
|
||||
case "http":
|
||||
u.Host += ":80"
|
||||
}
|
||||
}
|
||||
return u.String()
|
||||
}
|
||||
|
||||
// conflictURL is conflictString for URL-typed keys: both sides are
|
||||
// normalized via canonicalURL before comparison.
|
||||
func conflictURL(key, got string) conflictCheck {
|
||||
return conflictCheck{
|
||||
key: key,
|
||||
check: func(pol *mdm.Policy) bool {
|
||||
if got == "" {
|
||||
return true
|
||||
}
|
||||
want, ok := pol.GetString(key)
|
||||
return ok && canonicalURL(want) == canonicalURL(got)
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// conflictString builds a conflictCheck for a string MDM key. An empty
|
||||
// `got` is treated as "field not set" (no override requested); otherwise
|
||||
// the check returns true only when the policy contains the key and its
|
||||
@@ -256,7 +288,7 @@ func mdmManagedFieldConflicts(msg *proto.SetConfigRequest, policy *mdm.Policy) [
|
||||
}
|
||||
|
||||
return resolveConflicts(policy, []conflictCheck{
|
||||
conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
|
||||
conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
|
||||
conflictString(mdm.KeyPreSharedKey, pskGot),
|
||||
conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
|
||||
conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),
|
||||
@@ -377,7 +409,7 @@ func loginRequestMDMConflicts(msg *proto.LoginRequest, policy *mdm.Policy) []str
|
||||
}
|
||||
|
||||
return resolveConflicts(policy, []conflictCheck{
|
||||
conflictString(mdm.KeyManagementURL, msg.ManagementUrl),
|
||||
conflictURL(mdm.KeyManagementURL, msg.ManagementUrl),
|
||||
conflictString(mdm.KeyPreSharedKey, pskGot),
|
||||
conflictBool(mdm.KeyRosenpassEnabled, msg.RosenpassEnabled),
|
||||
conflictBool(mdm.KeyRosenpassPermissive, msg.RosenpassPermissive),
|
||||
|
||||
@@ -181,6 +181,43 @@ func TestSetConfig_MDMAllow_NonManagedFields(t *testing.T) {
|
||||
require.NotNil(t, resp)
|
||||
}
|
||||
|
||||
// TestSetConfig_MDMAllow_ManagementURLPortNormalized covers the
|
||||
// regression from discussion #6483: MDM URL without explicit port vs
|
||||
// UI echo with the parseURL-appended default port must be treated as
|
||||
// a no-op echo, not a conflict.
|
||||
func TestSetConfig_MDMAllow_ManagementURLPortNormalized(t *testing.T) {
|
||||
tests := []struct {
|
||||
name string
|
||||
mdmURL string
|
||||
submitURL string
|
||||
}{
|
||||
{"policy_no_port_submit_with_443", "https://netbird.corp.example", "https://netbird.corp.example:443"},
|
||||
{"policy_with_443_submit_no_port", "https://netbird.corp.example:443", "https://netbird.corp.example"},
|
||||
{"http_policy_no_port_submit_with_80", "http://netbird.corp.example", "http://netbird.corp.example:80"},
|
||||
}
|
||||
|
||||
for _, tc := range tests {
|
||||
t.Run(tc.name, func(t *testing.T) {
|
||||
withMDMPolicy(t, mdm.NewPolicy(map[string]any{
|
||||
mdm.KeyManagementURL: tc.mdmURL,
|
||||
}))
|
||||
|
||||
s, ctx, profName, username, _ := setupServerWithProfile(t)
|
||||
|
||||
rosenpassEnabled := true
|
||||
resp, err := s.SetConfig(ctx, &proto.SetConfigRequest{
|
||||
ProfileName: profName,
|
||||
Username: username,
|
||||
ManagementUrl: tc.submitURL,
|
||||
RosenpassEnabled: &rosenpassEnabled,
|
||||
})
|
||||
|
||||
require.NoError(t, err, "port-normalized URL echo must not trip MDM conflict gate")
|
||||
require.NotNil(t, resp)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestSetConfig_MDMEmpty_NoEnforcement(t *testing.T) {
|
||||
// No MDM policy active: any field can be written.
|
||||
withMDMPolicy(t, mdm.NewPolicy(nil))
|
||||
|
||||
@@ -7,7 +7,7 @@ import (
|
||||
"os"
|
||||
"slices"
|
||||
|
||||
"github.com/shirou/gopsutil/v3/process"
|
||||
"github.com/shirou/gopsutil/v4/process"
|
||||
)
|
||||
|
||||
// getRunningProcesses returns a list of running process paths. The context bounds the work:
|
||||
|
||||
@@ -4,7 +4,7 @@ import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/shirou/gopsutil/v3/process"
|
||||
"github.com/shirou/gopsutil/v4/process"
|
||||
)
|
||||
|
||||
func Benchmark_getRunningProcesses(b *testing.B) {
|
||||
|
||||
2
go.mod
2
go.mod
@@ -104,6 +104,7 @@ require (
|
||||
github.com/redis/go-redis/v9 v9.7.3
|
||||
github.com/rs/xid v1.3.0
|
||||
github.com/shirou/gopsutil/v3 v3.24.4
|
||||
github.com/shirou/gopsutil/v4 v4.25.8
|
||||
github.com/skratchdot/open-golang v0.0.0-20200116055534-eef842397966
|
||||
github.com/songgao/water v0.0.0-20200317203138-2b4b6d7c09d8
|
||||
github.com/stretchr/testify v1.11.1
|
||||
@@ -308,7 +309,6 @@ require (
|
||||
github.com/russellhaering/goxmldsig v1.6.0 // indirect
|
||||
github.com/ryanuber/go-glob v1.0.0 // indirect
|
||||
github.com/rymdport/portal v0.4.2 // indirect
|
||||
github.com/shirou/gopsutil/v4 v4.25.8 // indirect
|
||||
github.com/shoenig/go-m1cpu v0.2.1 // indirect
|
||||
github.com/shopspring/decimal v1.4.0 // indirect
|
||||
github.com/spf13/cast v1.7.0 // indirect
|
||||
|
||||
@@ -30,11 +30,16 @@ type RelayTrack struct {
|
||||
relayClient *Client
|
||||
err error
|
||||
created time.Time
|
||||
// ready is closed once the dial started by openConnVia finishes (relayClient
|
||||
// or err is set). Callers reusing a track wait on this instead of the track
|
||||
// lock, so the dial never runs under rt.Lock.
|
||||
ready chan struct{}
|
||||
}
|
||||
|
||||
func NewRelayTrack() *RelayTrack {
|
||||
return &RelayTrack{
|
||||
created: time.Now(),
|
||||
ready: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -326,34 +331,24 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
// check if already has a connection to the desired relay server
|
||||
m.relayClientsMutex.RLock()
|
||||
rt, ok := m.relayClients[serverAddress]
|
||||
if ok {
|
||||
rt.RLock()
|
||||
m.relayClientsMutex.RUnlock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
m.relayClientsMutex.RUnlock()
|
||||
if ok {
|
||||
return m.openConnOnTrack(ctx, rt, peerKey)
|
||||
}
|
||||
|
||||
// if not, establish a new connection but check it again (because changed the lock type) before starting the
|
||||
// connection
|
||||
m.relayClientsMutex.Lock()
|
||||
rt, ok = m.relayClients[serverAddress]
|
||||
if ok {
|
||||
rt.RLock()
|
||||
m.relayClientsMutex.Unlock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
return m.openConnOnTrack(ctx, rt, peerKey)
|
||||
}
|
||||
|
||||
// create a new relay client and store it in the relayClients map
|
||||
// Publish the track and release the map lock BEFORE dialing, so the dial does
|
||||
// not run under rt.Lock (which would block RelayStates and the cleanup loop
|
||||
// for the full dial). Concurrent callers find this track and wait on rt.ready.
|
||||
rt = NewRelayTrack()
|
||||
rt.Lock()
|
||||
m.relayClients[serverAddress] = rt
|
||||
m.relayClientsMutex.Unlock()
|
||||
|
||||
@@ -361,8 +356,10 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
relayClient.SetTransportFallback(m.transportFallback)
|
||||
err := relayClient.Connect(m.ctx)
|
||||
if err != nil {
|
||||
rt.Lock()
|
||||
rt.err = err
|
||||
rt.Unlock()
|
||||
close(rt.ready)
|
||||
m.relayClientsMutex.Lock()
|
||||
delete(m.relayClients, serverAddress)
|
||||
m.relayClientsMutex.Unlock()
|
||||
@@ -370,14 +367,34 @@ func (m *Manager) openConnVia(ctx context.Context, serverAddress, peerKey string
|
||||
}
|
||||
// if connection closed then delete the relay client from the list
|
||||
relayClient.SetOnDisconnectListener(m.onServerDisconnected)
|
||||
rt.Lock()
|
||||
rt.relayClient = relayClient
|
||||
rt.Unlock()
|
||||
close(rt.ready)
|
||||
|
||||
conn, err := relayClient.OpenConn(ctx, peerKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
return relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
|
||||
// openConnOnTrack opens a peer connection through an existing relay track,
|
||||
// waiting for the dial started by another openConnVia call to finish. It waits
|
||||
// on rt.ready rather than the track lock, so it neither holds nor contends the
|
||||
// track lock across the dial.
|
||||
func (m *Manager) openConnOnTrack(ctx context.Context, rt *RelayTrack, peerKey string) (net.Conn, error) {
|
||||
select {
|
||||
case <-rt.ready:
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
}
|
||||
return conn, nil
|
||||
|
||||
rt.RLock()
|
||||
defer rt.RUnlock()
|
||||
if rt.err != nil {
|
||||
return nil, rt.err
|
||||
}
|
||||
if rt.relayClient == nil {
|
||||
return nil, ErrRelayClientNotConnected
|
||||
}
|
||||
return rt.relayClient.OpenConn(ctx, peerKey)
|
||||
}
|
||||
|
||||
func (m *Manager) onServerConnected() {
|
||||
@@ -476,6 +493,13 @@ func (m *Manager) cleanUpUnusedRelays() {
|
||||
continue
|
||||
}
|
||||
|
||||
// dial still in progress (openConnVia publishes the track before Connect
|
||||
// completes and no longer holds rt.Lock during it), nothing to clean up.
|
||||
if rt.relayClient == nil {
|
||||
rt.Unlock()
|
||||
continue
|
||||
}
|
||||
|
||||
if time.Since(rt.created) <= m.keepUnusedServerTime {
|
||||
rt.Unlock()
|
||||
continue
|
||||
|
||||
60
shared/relay/client/manager_cleanup_test.go
Normal file
60
shared/relay/client/manager_cleanup_test.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net/netip"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial drives a real, hanging foreign
|
||||
// relay dial and asserts cleanUpUnusedRelays does not stall behind it.
|
||||
func TestCleanUpUnusedRelays_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
serverAddr := stallingRelayListener(t)
|
||||
|
||||
mCtx, mCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(mCancel)
|
||||
|
||||
m := NewManager(mCtx, nil, "alice", 1280)
|
||||
|
||||
dialDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(dialDone)
|
||||
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
|
||||
}()
|
||||
|
||||
// The track appears in the map once the dial is in flight.
|
||||
require.Eventually(t, func() bool {
|
||||
m.relayClientsMutex.RLock()
|
||||
defer m.relayClientsMutex.RUnlock()
|
||||
_, ok := m.relayClients[serverAddr]
|
||||
return ok
|
||||
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
|
||||
|
||||
cleanupDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(cleanupDone)
|
||||
m.cleanUpUnusedRelays()
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-cleanupDone:
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("cleanUpUnusedRelays blocked on an in-progress relay dial while holding the relay map lock")
|
||||
}
|
||||
|
||||
m.relayClientsMutex.RLock()
|
||||
_, stillTracked := m.relayClients[serverAddr]
|
||||
m.relayClientsMutex.RUnlock()
|
||||
require.True(t, stillTracked, "an in-progress relay dial must not be evicted by cleanup")
|
||||
|
||||
// Release the hanging dial so the goroutine can exit cleanly.
|
||||
mCancel()
|
||||
select {
|
||||
case <-dialDone:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("openConnVia did not return after context cancellation")
|
||||
}
|
||||
}
|
||||
91
shared/relay/client/manager_relaystates_test.go
Normal file
91
shared/relay/client/manager_relaystates_test.go
Normal file
@@ -0,0 +1,91 @@
|
||||
package client
|
||||
|
||||
import (
|
||||
"context"
|
||||
"net"
|
||||
"net/netip"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/require"
|
||||
)
|
||||
|
||||
// stallingRelayListener accepts TCP connections and holds them open without ever
|
||||
// responding, so a relay handshake dialed against it blocks until its context is
|
||||
// cancelled. It returns the "rel://host:port" URL to dial.
|
||||
func stallingRelayListener(t *testing.T) string {
|
||||
t.Helper()
|
||||
|
||||
ln, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
|
||||
var mu sync.Mutex
|
||||
var conns []net.Conn
|
||||
go func() {
|
||||
for {
|
||||
c, err := ln.Accept()
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
mu.Lock()
|
||||
conns = append(conns, c)
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
t.Cleanup(func() {
|
||||
_ = ln.Close()
|
||||
mu.Lock()
|
||||
for _, c := range conns {
|
||||
_ = c.Close()
|
||||
}
|
||||
mu.Unlock()
|
||||
})
|
||||
|
||||
return "rel://" + ln.Addr().String()
|
||||
}
|
||||
|
||||
// TestRelayStates_DoesNotBlockOnRealHangingDial is a regression test for
|
||||
// RelayStates() called by a "status -d command" hanging behind an in-progress
|
||||
// relay dial.
|
||||
func TestRelayStates_DoesNotBlockOnRealHangingDial(t *testing.T) {
|
||||
serverAddr := stallingRelayListener(t)
|
||||
|
||||
mCtx, mCancel := context.WithCancel(context.Background())
|
||||
t.Cleanup(mCancel)
|
||||
|
||||
m := NewManager(mCtx, nil, "alice", 1280)
|
||||
|
||||
dialDone := make(chan struct{})
|
||||
go func() {
|
||||
defer close(dialDone)
|
||||
_, _ = m.openConnVia(mCtx, serverAddr, "peerKey", netip.Addr{})
|
||||
}()
|
||||
|
||||
require.Eventually(t, func() bool {
|
||||
m.relayClientsMutex.RLock()
|
||||
defer m.relayClientsMutex.RUnlock()
|
||||
_, ok := m.relayClients[serverAddr]
|
||||
return ok
|
||||
}, 5*time.Second, 5*time.Millisecond, "relay dial did not start")
|
||||
|
||||
done := make(chan []RelayConnState, 1)
|
||||
go func() {
|
||||
done <- m.RelayStates()
|
||||
}()
|
||||
|
||||
select {
|
||||
case states := <-done:
|
||||
require.Empty(t, states, "a relay still being dialed carries no state and must be omitted")
|
||||
case <-time.After(2 * time.Second):
|
||||
t.Fatal("RelayStates blocked on a foreign relay whose Connect() is in progress")
|
||||
}
|
||||
|
||||
// Release the hanging dial so the goroutine can exit cleanly.
|
||||
mCancel()
|
||||
select {
|
||||
case <-dialDone:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("openConnVia did not return after context cancellation")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user