Skip to content

Commit b3f39fe

Browse files
spikecurtisspikecurtis
committed
integration test improvements
1 parent 619a50e commit b3f39fe

File tree

4 files changed

+352
-235
lines changed

4 files changed

+352
-235
lines changed

‎tailnet/test/integration/integration.go

Lines changed: 153 additions & 65 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@ import (
1919
"sync"
2020
"sync/atomic"
2121
"syscall"
22+
"tailscale.com/net/packet"
23+
"tailscale.com/wgengine/capture"
2224
"testing"
2325
"time"
2426

@@ -54,35 +56,36 @@ type Client struct {
5456
ID uuid.UUID
5557
ListenPort uint16
5658
ShouldRunTests bool
59+
TunnelSrc bool
5760
}
5861

5962
var Client1 = Client{
6063
Number: ClientNumber1,
6164
ID: uuid.MustParse("00000000-0000-0000-0000-000000000001"),
6265
ListenPort: client1Port,
6366
ShouldRunTests: true,
67+
TunnelSrc: true,
6468
}
6569

6670
var Client2 = Client{
6771
Number: ClientNumber2,
6872
ID: uuid.MustParse("00000000-0000-0000-0000-000000000002"),
6973
ListenPort: client2Port,
7074
ShouldRunTests: false,
75+
TunnelSrc: false,
7176
}
7277

7378
type TestTopology struct {
7479
Name string
75-
// SetupNetworking creates interfaces and network namespaces for the test.
76-
// The most simple implementation is NetworkSetupDefault, which only creates
77-
// a network namespace shared for all tests.
78-
SetupNetworking func(t *testing.T, logger slog.Logger) TestNetworking
80+
81+
NetworkingProvider NetworkingProvider
7982

8083
// Server is the server starter for the test. It is executed in the server
8184
// subprocess.
8285
Server ServerStarter
83-
// StartClient gets called in each client subprocess. It's expected to
86+
// ClientStarter.StartClient gets called in each client subprocess. It's expected to
8487
// create the tailnet.Conn and ensure connectivity to it's peer.
85-
StartClient func(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me Client, peer Client) *tailnet.Conn
88+
ClientStarter ClientStarter
8689

8790
// RunTests is the main test function. It's called in each of the client
8891
// subprocesses. If tests can only run once, they should check the client ID
@@ -97,6 +100,17 @@ type ServerStarter interface {
97100
StartServer(t *testing.T, logger slog.Logger, listenAddr string)
98101
}
99102

103+
type NetworkingProvider interface {
104+
// SetupNetworking creates interfaces and network namespaces for the test.
105+
// The most simple implementation is NetworkSetupDefault, which only creates
106+
// a network namespace shared for all tests.
107+
SetupNetworking(t *testing.T, logger slog.Logger) TestNetworking
108+
}
109+
110+
type ClientStarter interface {
111+
StartClient(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me Client, peer Client) *tailnet.Conn
112+
}
113+
100114
type SimpleServerOptions struct {
101115
// FailUpgradeDERP will make the DERP server fail to handle the initial DERP
102116
// upgrade in a way that causes the client to fallback to
@@ -369,77 +383,106 @@ http {
369383
_, _ = ExecBackground(t, "server.nginx", nil, "nginx", []string{"-c", cfgPath})
370384
}
371385

372-
// StartClientDERP creates a client connection to the server for coordination
373-
// and creates a tailnet.Conn which will only use DERP to connect to the peer.
374-
func StartClientDERP(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me, peer Client) *tailnet.Conn {
375-
return startClientOptions(t, logger, serverURL, me, peer, &tailnet.Options{
376-
Addresses: []netip.Prefix{tailnet.TailscaleServicePrefix.PrefixFromUUID(me.ID)},
377-
DERPMap: derpMap,
378-
BlockEndpoints: true,
379-
Logger: logger,
380-
DERPForceWebSockets: false,
381-
ListenPort: me.ListenPort,
382-
// These tests don't have internet connection, so we need to force
383-
// magicsock to do anything.
384-
ForceNetworkUp: true,
385-
})
386+
type BasicClientStarter struct {
387+
BlockEndpoints bool
388+
DERPForceWebsockets bool
389+
// WaitForConnection means wait for (any) peer connection before returning from StartClient
390+
WaitForConnection bool
391+
// WaitForConnection means wait for a direct peer connection before returning from StartClient
392+
WaitForDirect bool
393+
// Service is a network service (e.g. an echo server) to start on the client. If Wait* is set, the service is
394+
// started prior to waiting.
395+
Service NetworkService
396+
LogPackets bool
386397
}
387398

388-
// StartClientDERPWebSockets does the same thing as StartClientDERP but will
389-
// only use DERP WebSocket fallback.
390-
func StartClientDERPWebSockets(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me, peer Client) *tailnet.Conn {
391-
return startClientOptions(t, logger, serverURL, me, peer, &tailnet.Options{
392-
Addresses: []netip.Prefix{tailnet.TailscaleServicePrefix.PrefixFromUUID(me.ID)},
393-
DERPMap: derpMap,
394-
BlockEndpoints: true,
395-
Logger: logger,
396-
DERPForceWebSockets: true,
397-
ListenPort: me.ListenPort,
398-
// These tests don't have internet connection, so we need to force
399-
// magicsock to do anything.
400-
ForceNetworkUp: true,
401-
})
399+
type NetworkService interface {
400+
StartService(t *testing.T, logger slog.Logger, conn *tailnet.Conn)
402401
}
403402

404-
// StartClientDirect does the same thing as StartClientDERP but disables
405-
// BlockEndpoints (which enables Direct connections), and waits for a direct
406-
// connection to be established between the two peers.
407-
func StartClientDirect(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me, peer Client) *tailnet.Conn {
403+
func (b BasicClientStarter) StartClient(t *testing.T, logger slog.Logger, serverURL *url.URL, derpMap *tailcfg.DERPMap, me, peer Client) *tailnet.Conn {
404+
var hook capture.Callback
405+
if b.LogPackets {
406+
pktLogger := packetLogger{logger}
407+
hook = pktLogger.LogPacket
408+
}
408409
conn := startClientOptions(t, logger, serverURL, me, peer, &tailnet.Options{
409410
Addresses: []netip.Prefix{tailnet.TailscaleServicePrefix.PrefixFromUUID(me.ID)},
410411
DERPMap: derpMap,
411-
BlockEndpoints: false,
412+
BlockEndpoints: b.BlockEndpoints,
412413
Logger: logger,
413-
DERPForceWebSockets: true,
414+
DERPForceWebSockets: b.DERPForceWebsockets,
414415
ListenPort: me.ListenPort,
415416
// These tests don't have internet connection, so we need to force
416417
// magicsock to do anything.
417418
ForceNetworkUp: true,
419+
CaptureHook: hook,
418420
})
419421

420-
// Wait for direct connection to be established.
421-
peerIP := tailnet.TailscaleServicePrefix.AddrFromUUID(peer.ID)
422-
require.Eventually(t, func() bool {
423-
t.Log("attempting ping to peer to judge direct connection")
424-
ctx := testutil.Context(t, testutil.WaitShort)
425-
_, p2p, pong, err := conn.Ping(ctx, peerIP)
426-
if err != nil {
427-
t.Logf("ping failed: %v", err)
428-
return false
429-
}
430-
if !p2p {
431-
t.Log("ping succeeded, but not direct yet")
432-
return false
433-
}
434-
t.Logf("ping succeeded, direct connection established via %s", pong.Endpoint)
435-
return true
436-
}, testutil.WaitLong, testutil.IntervalMedium)
422+
if b.Service != nil {
423+
b.Service.StartService(t, logger, conn)
424+
}
425+
426+
if b.WaitForConnection || b.WaitForDirect {
427+
// Wait for connection to be established.
428+
peerIP := tailnet.TailscaleServicePrefix.AddrFromUUID(peer.ID)
429+
require.Eventually(t, func() bool {
430+
t.Log("attempting ping to peer to judge direct connection")
431+
ctx := testutil.Context(t, testutil.WaitShort)
432+
_, p2p, pong, err := conn.Ping(ctx, peerIP)
433+
if err != nil {
434+
t.Logf("ping failed: %v", err)
435+
return false
436+
}
437+
if !p2p && b.WaitForDirect {
438+
t.Log("ping succeeded, but not direct yet")
439+
return false
440+
}
441+
t.Logf("ping succeeded, p2p=%t, endpoint=%s", p2p, pong.Endpoint)
442+
return true
443+
}, testutil.WaitLong, testutil.IntervalMedium)
444+
}
437445

438446
return conn
439447
}
440448

441-
type ClientStarter struct {
442-
Options *tailnet.Options
449+
const EchoPort = 2381
450+
451+
type UDPEchoService struct{}
452+
453+
func (UDPEchoService) StartService(t *testing.T, logger slog.Logger, _ *tailnet.Conn) {
454+
// tailnet doesn't handle UDP connections "in-process" the way we do for TCP, so we need to listen in the OS,
455+
// and tailnet will forward
456+
//packets.
457+
l, err := net.ListenUDP("udp", &net.UDPAddr{
458+
IP: net.IPv6zero, // all interfaces
459+
Port: EchoPort,
460+
})
461+
require.NoError(t, err)
462+
logger.Info(context.Background(), "started UDPEcho server")
463+
t.Cleanup(func() {
464+
lCloseErr := l.Close()
465+
if lCloseErr != nil {
466+
t.Logf("error closing UDPEcho listener: %v", lCloseErr)
467+
}
468+
})
469+
go func() {
470+
buf := make([]byte, 1500)
471+
for {
472+
n, remote, readErr := l.ReadFromUDP(buf)
473+
if readErr != nil {
474+
logger.Info(context.Background(), "error reading UDPEcho listener", slog.Error(readErr))
475+
return
476+
}
477+
logger.Info(context.Background(), "received UDP packet",
478+
slog.F("len", n), slog.F("remote", remote))
479+
n, writeErr := l.WriteToUDP(buf[:n], remote)
480+
if writeErr != nil {
481+
logger.Info(context.Background(), "error writing UDPEcho listener", slog.Error(writeErr))
482+
return
483+
}
484+
}
485+
}()
443486
}
444487

445488
func startClientOptions(t *testing.T, logger slog.Logger, serverURL *url.URL, me, peer Client, options *tailnet.Options) *tailnet.Conn {
@@ -467,9 +510,16 @@ func startClientOptions(t *testing.T, logger slog.Logger, serverURL *url.URL, me
467510
_ = conn.Close()
468511
})
469512

470-
ctrl := tailnet.NewTunnelSrcCoordController(logger, conn)
471-
ctrl.AddDestination(peer.ID)
472-
coordination := ctrl.New(coord)
513+
var coordination tailnet.CloserWaiter
514+
if me.TunnelSrc {
515+
ctrl := tailnet.NewTunnelSrcCoordController(logger, conn)
516+
ctrl.AddDestination(peer.ID)
517+
coordination = ctrl.New(coord)
518+
} else {
519+
// use the "Agent" controller so that we act as a tunnel destination and send "ReadyForHandshake" acks.
520+
ctrl := tailnet.NewAgentCoordinationController(logger, conn)
521+
coordination = ctrl.New(coord)
522+
}
473523
t.Cleanup(func() {
474524
cctx, cancel := context.WithTimeout(context.Background(), testutil.WaitShort)
475525
defer cancel()
@@ -492,11 +542,17 @@ func basicDERPMap(serverURLStr string) (*tailcfg.DERPMap, error) {
492542
}
493543

494544
hostname := serverURL.Hostname()
495-
ipv4 := ""
545+
ipv4 := "none"
546+
ipv6 := "none"
496547
ip, err := netip.ParseAddr(hostname)
497548
if err == nil {
498549
hostname = ""
499-
ipv4 = ip.String()
550+
if ip.Is4() {
551+
ipv4 = ip.String()
552+
}
553+
if ip.Is6() {
554+
ipv6 = ip.String()
555+
}
500556
}
501557

502558
return &tailcfg.DERPMap{
@@ -511,7 +567,7 @@ func basicDERPMap(serverURLStr string) (*tailcfg.DERPMap, error) {
511567
RegionID: 1,
512568
HostName: hostname,
513569
IPv4: ipv4,
514-
IPv6: "none",
570+
IPv6: ipv6,
515571
DERPPort: port,
516572
STUNPort: -1,
517573
ForceHTTP: true,
@@ -648,3 +704,35 @@ func (w *testWriter) Flush() {
648704
}
649705
w.capturedLines = nil
650706
}
707+
708+
type packetLogger struct {
709+
l slog.Logger
710+
}
711+
712+
func (p packetLogger) LogPacket(path capture.Path, when time.Time, pkt []byte, _ packet.CaptureMeta) {
713+
q := new(packet.Parsed)
714+
q.Decode(pkt)
715+
p.l.Info(context.Background(), "Packet",
716+
slog.F("path", pathString(path)),
717+
slog.F("when", when),
718+
slog.F("decode", q.String()),
719+
slog.F("len", len(pkt)),
720+
)
721+
}
722+
723+
func pathString(path capture.Path) string {
724+
switch path {
725+
case capture.FromLocal:
726+
return "Local"
727+
case capture.FromPeer:
728+
return "Peer"
729+
case capture.SynthesizedToLocal:
730+
return "SynthesizedToLocal"
731+
case capture.SynthesizedToPeer:
732+
return "SynthesizedToPeer"
733+
case capture.PathDisco:
734+
return "Disco"
735+
default:
736+
return "<<UNKNOWN>>"
737+
}
738+
}

0 commit comments

Comments
 (0)