From d5062dacf0f8085ded1b7be9f8c0a57349baf71f Mon Sep 17 00:00:00 2001 From: Nate Brown Date: Wed, 30 Sep 2026 15:26:34 -0500 Subject: [PATCH] lighthouse: Never block in QueryServer, drop queries when the channel is full (#1910) --- connection_manager_test.go | 3 +++ examples/config.yml | 4 +-- lighthouse.go | 15 ++++++++--- lighthouse_test.go | 54 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 71 insertions(+), 5 deletions(-) diff --git a/connection_manager_test.go b/connection_manager_test.go index 6be63f7a..68b4dc01 100644 --- a/connection_manager_test.go +++ b/connection_manager_test.go @@ -7,6 +7,7 @@ import ( "testing" "time" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/overlay/overlaytest" @@ -21,6 +22,8 @@ func newTestLighthouse() *LightHouse { l: test.NewLogger(), addrMap: map[netip.Addr]*RemoteList{}, queryChan: make(chan netip.Addr, 10), + // Nothing drains queryChan here, so an 11th query is dropped. + queryDropped: metrics.NilCounter{}, } lighthouses := []netip.Addr{} staticList := map[netip.Addr]struct{}{} diff --git a/examples/config.yml b/examples/config.yml index 20a85883..0edb4dc6 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -410,8 +410,8 @@ logging: #try_interval: 100ms #retries: 20 - # query_buffer is the size of the buffer channel for querying lighthouses - #query_buffer: 64 + # query_buffer is the size of the buffer channel for querying lighthouses, drops are recorded in the lighthouse.query.dropped metric. + #query_buffer: 1024 # trigger_buffer is the size of the buffer channel for quickly sending handshakes # after receiving the response for lighthouse queries diff --git a/lighthouse.go b/lighthouse.go index 58037935..21937bcd 100644 --- a/lighthouse.go +++ b/lighthouse.go @@ -15,6 +15,7 @@ import ( "time" "github.com/gaissmai/bart" + "github.com/rcrowley/go-metrics" "github.com/slackhq/nebula/cert" "github.com/slackhq/nebula/config" "github.com/slackhq/nebula/header" @@ -81,6 +82,8 @@ type LightHouse struct { updateTrigger chan struct{} queryChan chan netip.Addr + // queryDropped counts queries QueryServer dropped because queryChan was full. + queryDropped metrics.Counter calculatedRemotes atomic.Pointer[bart.Table[[]*calculatedRemote]] // Maps VpnAddr to []*calculatedRemote @@ -116,7 +119,8 @@ func NewLightHouseFromConfig(ctx context.Context, l *slog.Logger, c *config.C, c nebulaPort: nebulaPort, punchy: p, updateTrigger: make(chan struct{}, 1), - queryChan: make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 64)), + queryChan: make(chan netip.Addr, c.GetUint32("handshakes.query_buffer", 1024)), + queryDropped: metrics.GetOrRegisterCounter("lighthouse.query.dropped", nil), l: l, } h.localAddrsFn = func(al *LocalAllowList) []netip.Addr { @@ -512,14 +516,19 @@ func (lh *LightHouse) Query(vpnAddr netip.Addr) *RemoteList { return nil } -// QueryServer is asynchronous so no reply should be expected +// QueryServer is asynchronous and non-blocking, no reply should be expected. +// If the query channel is full, the query is dropped. func (lh *LightHouse) QueryServer(vpnAddr netip.Addr) { // Don't put lighthouse addrs in the query channel because we can't query lighthouses about lighthouses if lh.amLighthouse || lh.IsLighthouseAddr(vpnAddr) { return } - lh.queryChan <- vpnAddr + select { + case lh.queryChan <- vpnAddr: + default: + lh.queryDropped.Inc(1) + } } func (lh *LightHouse) QueryCache(vpnAddrs []netip.Addr) *RemoteList { diff --git a/lighthouse_test.go b/lighthouse_test.go index 8fc30223..c39ffa16 100644 --- a/lighthouse_test.go +++ b/lighthouse_test.go @@ -6,7 +6,9 @@ import ( "errors" "fmt" "net/netip" + "sync" "testing" + "time" "github.com/gaissmai/bart" "github.com/slackhq/nebula/cert" @@ -795,3 +797,55 @@ func TestLightHouse_logLocalAddrsErr(t *testing.T) { lh.logLocalAddrsErr(errors.New("something else")) assert.Contains(t, out.String(), "level=WARN") } + +// heldEncWriter holds the query worker in its first GetHostInfo until release closes. +type heldEncWriter struct { + mockEncWriter + once sync.Once + held chan struct{} + release chan struct{} +} + +func (w *heldEncWriter) GetHostInfo(netip.Addr) *HostInfo { + w.once.Do(func() { close(w.held) }) + <-w.release + return nil +} + +// QueryServer never waits on the query worker, since the packet readers call it and nothing drains the channel at +// shutdown; a query that finds the channel full is dropped and counted. The worker is held on its first query so the +// channel fills. +func TestLighthouse_QueryServerDoesNotBlock(t *testing.T) { + l := test.NewLogger() + cs := testCertState(netip.MustParsePrefix("10.128.0.1/16")) + lh1 := "10.128.0.2" + c := config.NewC(l) + c.Settings["lighthouse"] = map[string]any{"hosts": []any{lh1}} + c.Settings["static_host_map"] = map[string]any{lh1: []any{"1.1.1.1:4242"}} + c.Settings["handshakes"] = map[string]any{"query_buffer": 1} + lh, err := NewLightHouseFromConfig(t.Context(), l, c, cs, nil, nil) + require.NoError(t, err) + w := &heldEncWriter{held: make(chan struct{}), release: make(chan struct{})} + lh.ifce = w + t.Cleanup(func() { close(w.release) }) + + lh.QueryServer(netip.MustParseAddr("10.128.0.10")) + select { + case <-w.held: + case <-time.After(5 * time.Second): + t.Fatal("the query worker never took the first query") + } + dropped := lh.queryDropped.Count() + done := make(chan struct{}) + go func() { + lh.QueryServer(netip.MustParseAddr("10.128.0.11")) + lh.QueryServer(netip.MustParseAddr("10.128.0.12")) + close(done) + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("QueryServer blocked on a full query channel") + } + assert.EqualValues(t, 1, lh.queryDropped.Count()-dropped) +}