mirror of
https://github.com/juanfont/headscale.git
synced 2026-10-28 09:16:50 +09:00
mapper: drop pending changes for nodes without a connection
Disconnected nodes keep their batcher entry, and every broadcast was rendered for them only for send to discard it.
This commit is contained in:
1 parent
5d21186f22
commit
fb015933ca
3 files changed
+44
-1
No files matched your search
@@ -8,6 +8,7 @@ import (
|
|||||||
"github.com/juanfont/headscale/hscontrol/types/change"
|
"github.com/juanfont/headscale/hscontrol/types/change"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
|
"tailscale.com/tailcfg"
|
||||||
)
|
)
|
||||||
|
|
||||||
// TestProcessBatchedChangesCoalescesWhenInFlight verifies that at most one
|
// TestProcessBatchedChangesCoalescesWhenInFlight verifies that at most one
|
||||||
@@ -20,6 +21,7 @@ func TestProcessBatchedChangesCoalescesWhenInFlight(t *testing.T) {
|
|||||||
|
|
||||||
id := types.NodeID(1)
|
id := types.NodeID(1)
|
||||||
nc := newMultiChannelNodeConn(id, nil)
|
nc := newMultiChannelNodeConn(id, nil)
|
||||||
|
nc.addConnection(makeConnectionEntry("conn-1", make(chan *tailcfg.MapResponse, 1)))
|
||||||
b.nodes.Store(id, nc)
|
b.nodes.Store(id, nc)
|
||||||
|
|
||||||
// A bundle is in flight: the new change must be retained, not queued.
|
// A bundle is in flight: the new change must be retained, not queued.
|
||||||
@@ -45,3 +47,25 @@ func TestProcessBatchedChangesCoalescesWhenInFlight(t *testing.T) {
|
|||||||
|
|
||||||
require.NotNil(t, b)
|
require.NotNil(t, b)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// TestProcessBatchedChangesSkipsNodesWithoutConnection drains a node that
|
||||||
|
// kept its batcher entry after its last stream closed. Rendering its changes
|
||||||
|
// is wasted, since send drops the result and its next stream opens with a
|
||||||
|
// full map, so no work may be queued for it.
|
||||||
|
func TestProcessBatchedChangesSkipsNodesWithoutConnection(t *testing.T) {
|
||||||
|
b := NewBatcher(50*time.Millisecond, 2, nil) // not started: no ticker, no workers
|
||||||
|
|
||||||
|
id := types.NodeID(1)
|
||||||
|
nc := newMultiChannelNodeConn(id, nil)
|
||||||
|
b.nodes.Store(id, nc)
|
||||||
|
|
||||||
|
nc.appendPending(change.PolicyChange())
|
||||||
|
b.processBatchedChanges()
|
||||||
|
|
||||||
|
nc.pendingMu.Lock()
|
||||||
|
pending := len(nc.pending)
|
||||||
|
nc.pendingMu.Unlock()
|
||||||
|
assert.Equal(t, 0, pending, "a node without a connection must not accumulate pending changes")
|
||||||
|
assert.Empty(t, b.workCh, "no work may be queued for a node without a connection")
|
||||||
|
assert.False(t, nc.inFlight.Load())
|
||||||
|
}
|
||||||
@@ -745,7 +745,7 @@ func (b *Batcher) processBatchedChanges() {
|
|||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|
||||||
pending := nc.drainPending()
|
pending := nc.takePending()
|
||||||
if len(pending) == 0 {
|
if len(pending) == 0 {
|
||||||
return true
|
return true
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -319,6 +319,25 @@ func (mc *multiChannelNodeConn) drainPending() []change.Change {
|
|||||||
return p
|
return p
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// takePending drains the pending changes for rendering. A node without a
|
||||||
|
// connection gets none and its changes are dropped: send would discard what
|
||||||
|
// they render, and its next connection opens with a full map rendered after
|
||||||
|
// it registered. The read lock keeps that registration out of the gap between
|
||||||
|
// the check and the drain.
|
||||||
|
func (mc *multiChannelNodeConn) takePending() []change.Change {
|
||||||
|
mc.mutex.RLock()
|
||||||
|
defer mc.mutex.RUnlock()
|
||||||
|
|
||||||
|
p := mc.drainPending()
|
||||||
|
if len(mc.connections) == 0 {
|
||||||
|
mapperChangesDropped.WithLabelValues("no_connection").Add(float64(len(p)))
|
||||||
|
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
|
||||||
|
return p
|
||||||
|
}
|
||||||
|
|
||||||
// send broadcasts data to all active connections for the node.
|
// send broadcasts data to all active connections for the node.
|
||||||
//
|
//
|
||||||
// To avoid holding the write lock during potentially slow sends (each stale
|
// To avoid holding the write lock during potentially slow sends (each stale
|
||||||
|
|||||||
Reference in new issue
Block a user