mirror of
https://github.com/juanfont/headscale.git
synced 2026-07-08 00:50:20 +09:00
mapper,policy: add reconnect-storm and lock-concurrency regression tests
TestInitialMapNotStarvedByReconnectStorm reproduces the #3346 stall; TestPolicyManagerConcurrentReads guards the RLock cache access under -race. Updates #3346
This commit is contained in:
Kristoffer Dalby
@@ -0,0 +1,189 @@
|
||||
//go:build !race
|
||||
|
||||
// This is a timing-sensitive performance regression test; the race detector's
|
||||
// ~10x slowdown makes its wall-clock assertion meaningless, so it is excluded
|
||||
// from -race builds. The concurrency correctness of the policy lock change it
|
||||
// guards is covered under -race by TestPolicyManagerConcurrentReads in
|
||||
// hscontrol/policy/v2.
|
||||
|
||||
package mapper
|
||||
|
||||
import (
|
||||
"net/netip"
|
||||
"runtime"
|
||||
"slices"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/juanfont/headscale/hscontrol/db"
|
||||
"github.com/juanfont/headscale/hscontrol/derp"
|
||||
"github.com/juanfont/headscale/hscontrol/state"
|
||||
"github.com/juanfont/headscale/hscontrol/types"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"tailscale.com/tailcfg"
|
||||
)
|
||||
|
||||
// setupStormBatcher builds a real state+batcher with production-default
|
||||
// NodeStore batching so the reconnect-storm contention is realistic. It mirrors
|
||||
// setupBatcherWithTestData but lets the test control BatcherWorkers and the
|
||||
// policy.
|
||||
func setupStormBatcher(tb testing.TB, nodeCount, workers int, policy string) (*TestData, func()) {
|
||||
tb.Helper()
|
||||
|
||||
tmpDir := tb.TempDir()
|
||||
prefixV4 := netip.MustParsePrefix("100.64.0.0/10")
|
||||
prefixV6 := netip.MustParsePrefix("fd7a:115c:a1e0::/48")
|
||||
|
||||
cfg := &types.Config{
|
||||
Database: types.DatabaseConfig{
|
||||
Type: types.DatabaseSqlite,
|
||||
Sqlite: types.SqliteConfig{Path: tmpDir + "/headscale_test.db"},
|
||||
},
|
||||
PrefixV4: &prefixV4,
|
||||
PrefixV6: &prefixV6,
|
||||
IPAllocation: types.IPAllocationStrategySequential,
|
||||
BaseDomain: "headscale.test",
|
||||
Policy: types.PolicyConfig{Mode: types.PolicyModeDB},
|
||||
DERP: types.DERPConfig{
|
||||
ServerEnabled: false,
|
||||
DERPMap: &tailcfg.DERPMap{
|
||||
Regions: map[int]*tailcfg.DERPRegion{999: {RegionID: 999}},
|
||||
},
|
||||
},
|
||||
Tuning: types.Tuning{
|
||||
BatchChangeDelay: 10 * time.Millisecond,
|
||||
BatcherWorkers: workers,
|
||||
// Production defaults: coalesce writes so the storm is not
|
||||
// exaggerated by an unrealistically small NodeStore batch.
|
||||
NodeStoreBatchSize: 100,
|
||||
NodeStoreBatchTimeout: 500 * time.Millisecond,
|
||||
},
|
||||
}
|
||||
|
||||
database, err := db.NewHeadscaleDatabase(cfg)
|
||||
require.NoError(tb, err)
|
||||
|
||||
users := database.CreateUsersForTest(1, "testuser")
|
||||
dbNodes := database.CreateRegisteredNodesForTest(users[0], nodeCount, "node")
|
||||
|
||||
allNodes := make([]node, 0, nodeCount)
|
||||
for i := range dbNodes {
|
||||
allNodes = append(allNodes, node{
|
||||
n: dbNodes[i],
|
||||
ch: make(chan *tailcfg.MapResponse, normalBufferSize),
|
||||
})
|
||||
}
|
||||
|
||||
st, err := state.NewState(cfg)
|
||||
require.NoError(tb, err)
|
||||
|
||||
derpMap, err := derp.GetDERPMap(cfg.DERP)
|
||||
require.NoError(tb, err)
|
||||
st.SetDERPMap(derpMap)
|
||||
|
||||
_, err = st.SetPolicy([]byte(policy))
|
||||
require.NoError(tb, err)
|
||||
|
||||
batcher := wrapBatcherForTest(NewBatcherAndMapper(cfg, st), st)
|
||||
batcher.Start()
|
||||
|
||||
td := &TestData{
|
||||
Database: database,
|
||||
Users: users,
|
||||
Nodes: allNodes,
|
||||
State: st,
|
||||
Config: cfg,
|
||||
Batcher: batcher,
|
||||
}
|
||||
|
||||
return td, func() {
|
||||
batcher.Close()
|
||||
st.Close()
|
||||
database.Close()
|
||||
}
|
||||
}
|
||||
|
||||
// TestInitialMapNotStarvedByReconnectStorm reproduces juanfont/headscale#3346.
|
||||
//
|
||||
// When every node redials at once (e.g. after a server upgrade restart), each
|
||||
// connection writes the NodeStore (UpdateNodeFromMapRequest + Connect) and the
|
||||
// batcher generates its initial map. All of that reads the policy through the
|
||||
// PolicyManager. Before the fix the PolicyManager guarded every read with a
|
||||
// single exclusive mutex, so the NodeStore writer's O(n^2) BuildPeerMap and
|
||||
// every node's FilterForNode serialised against each other. On a per-node
|
||||
// filter policy (autogroup:self, via, relay grants) each hold is expensive, so
|
||||
// under the storm time-to-initial-map grew without bound.
|
||||
//
|
||||
// On the production server in #3346 this drove the batcher's per-node
|
||||
// total.duration from ~4s to ~76s; tailscale clients aborted the map POST
|
||||
// first and reported
|
||||
//
|
||||
// PollNetMap: Post ".../machine/map": unexpected EOF
|
||||
//
|
||||
// then redialled, feeding the storm so it never converged. An allow-all policy
|
||||
// does NOT reproduce this — BuildPeerMap is cheap there; the per-node filter
|
||||
// path is what makes it expensive, matching a real deployment's ACLs.
|
||||
//
|
||||
// The fix makes PolicyManager reads take a shared RLock so map generation runs
|
||||
// concurrently. AddNode blocks until the initial map is generated and handed to
|
||||
// the node channel, so its wall-clock duration is the time-to-initial-map the
|
||||
// client experiences. Without the fix this test's slowest node takes ~10s+ at
|
||||
// this scale (lock-bound, and more workers do not help); with it, generation
|
||||
// parallelises across workers and stays well within a client's patience.
|
||||
func TestInitialMapNotStarvedByReconnectStorm(t *testing.T) {
|
||||
if testing.Short() {
|
||||
t.Skip("timing-sensitive storm regression; skipped in -short")
|
||||
}
|
||||
|
||||
const (
|
||||
nodeCount = 300
|
||||
|
||||
// A per-node-filter policy: forces BuildPeerMap and FilterForNode onto
|
||||
// the slow path that recompiles filter rules per node, the same shape
|
||||
// as a real ACL using autogroup:self / via / relay grants.
|
||||
perNodeFilterPolicy = `{"acls":[{"action":"accept","src":["autogroup:member"],"dst":["autogroup:self:*"]}]}`
|
||||
|
||||
// Deliberately roomy so it passes on CI's few-core runners, where the
|
||||
// single-writer BuildPeerMap sets the floor (~10s) whatever the reads
|
||||
// do. It still trips on a hang or a return to the ~76s serialised
|
||||
// behaviour; the fine-grained concurrency is verified separately by
|
||||
// TestPolicyManagerConcurrentReads under -race.
|
||||
maxAcceptableLatency = 30 * time.Second
|
||||
)
|
||||
|
||||
// Use the real available parallelism, as production does.
|
||||
workers := runtime.NumCPU()
|
||||
|
||||
td, cleanup := setupStormBatcher(t, nodeCount, workers, perNodeFilterPolicy)
|
||||
defer cleanup()
|
||||
|
||||
latencies := make([]time.Duration, nodeCount)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
for i := range td.Nodes {
|
||||
wg.Go(func() {
|
||||
n := &td.Nodes[i]
|
||||
|
||||
start := time.Now()
|
||||
err := td.Batcher.AddNode(n.n.ID, n.ch, tailcfg.CapabilityVersion(100), nil)
|
||||
latencies[i] = time.Since(start)
|
||||
|
||||
assert.NoError(t, err) //nolint:testifylint // assert (not require) is correct off the test goroutine
|
||||
})
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
|
||||
slices.Sort(latencies)
|
||||
p50 := latencies[len(latencies)/2]
|
||||
p95 := latencies[len(latencies)*95/100]
|
||||
maxLatency := latencies[len(latencies)-1]
|
||||
t.Logf("initial-map latency over %d nodes (workers=%d): p50=%s p95=%s max=%s",
|
||||
nodeCount, workers, p50, p95, maxLatency)
|
||||
|
||||
require.Less(t, maxLatency, maxAcceptableLatency,
|
||||
"slowest initial map took %s: policy reads are serialising instead of running concurrently (issue #3346)", maxLatency)
|
||||
}
|
||||
@@ -0,0 +1,103 @@
|
||||
package v2
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"github.com/juanfont/headscale/hscontrol/types"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
// TestPolicyManagerConcurrentReads is the correctness guard for the #3346 fix:
|
||||
// PolicyManager read methods take a shared RLock and populate their per-node
|
||||
// caches (filterRulesMap, matchersForNodeMap) concurrently. This test hammers
|
||||
// those reads from many goroutines while a writer mutates the node set, so the
|
||||
// race detector catches any unsafe access to the shared caches or policy state.
|
||||
//
|
||||
// It uses an autogroup:self policy so reads take the per-node filter slow path
|
||||
// — the same path that made #3346's reconnect storm expensive — which is where
|
||||
// the lazy caches are written.
|
||||
func TestPolicyManagerConcurrentReads(t *testing.T) {
|
||||
users := types.Users{
|
||||
{Model: gorm.Model{ID: 1}, Name: "user1", Email: "user1@headscale.net"},
|
||||
{Model: gorm.Model{ID: 2}, Name: "user2", Email: "user2@headscale.net"},
|
||||
{Model: gorm.Model{ID: 3}, Name: "user3", Email: "user3@headscale.net"},
|
||||
}
|
||||
|
||||
policy := `{
|
||||
"acls": [
|
||||
{
|
||||
"action": "accept",
|
||||
"src": ["autogroup:member"],
|
||||
"dst": ["autogroup:self:*"]
|
||||
}
|
||||
]
|
||||
}`
|
||||
|
||||
const nodeCount = 60
|
||||
|
||||
nodes := make(types.Nodes, 0, nodeCount)
|
||||
for i := range nodeCount {
|
||||
n := node(
|
||||
fmt.Sprintf("node%d", i),
|
||||
fmt.Sprintf("100.64.0.%d", i+1),
|
||||
fmt.Sprintf("fd7a:115c:a1e0::%d", i+1),
|
||||
users[i%len(users)],
|
||||
)
|
||||
n.ID = types.NodeID(i + 1) //nolint:gosec // safe in test
|
||||
nodes = append(nodes, n)
|
||||
}
|
||||
|
||||
pm, err := NewPolicyManager([]byte(policy), users, nodes.ViewSlice())
|
||||
require.NoError(t, err)
|
||||
|
||||
const (
|
||||
readers = 16
|
||||
iterations = 60
|
||||
mutatorReloads = 30
|
||||
)
|
||||
|
||||
var wg sync.WaitGroup
|
||||
|
||||
// Concurrent readers exercise every converted RLock read path, including
|
||||
// the two lazily populated per-node caches. Assertions inside the
|
||||
// goroutines use assert (not require) so a failure does not call
|
||||
// t.FailNow from a non-test goroutine.
|
||||
for r := range readers {
|
||||
wg.Go(func() {
|
||||
for i := range iterations {
|
||||
nv := nodes[(r+i)%len(nodes)].View()
|
||||
|
||||
rules, err := pm.FilterForNode(nv)
|
||||
assert.NoError(t, err) //nolint:testifylint // assert (not require) is correct off the test goroutine
|
||||
assert.NotNil(t, rules)
|
||||
|
||||
_, err = pm.MatchersForNode(nv)
|
||||
assert.NoError(t, err) //nolint:testifylint // assert (not require) is correct off the test goroutine
|
||||
|
||||
pm.Filter()
|
||||
pm.NodeCapMap(nv.ID())
|
||||
|
||||
// BuildPeerMap is the O(n^2) writer-side read; exercise it
|
||||
// under RLock too, but not every iteration.
|
||||
if i%8 == 0 {
|
||||
assert.NotNil(t, pm.BuildPeerMap(nodes.ViewSlice()))
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// A writer repeatedly re-sets the node set, invalidating and racing the
|
||||
// caches the readers are populating.
|
||||
wg.Go(func() {
|
||||
for range mutatorReloads {
|
||||
_, err := pm.SetNodes(nodes.ViewSlice())
|
||||
assert.NoError(t, err) //nolint:testifylint // assert (not require) is correct off the test goroutine
|
||||
}
|
||||
})
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
Reference in New Issue
Block a user