From 2e5d08ec4f002f066cddb557687da089de201387 Mon Sep 17 00:00:00 2001 From: Joe Tsai Date: Mon, 28 Nov 2022 15:59:33 -0800 Subject: [PATCH] net/connstats: invert network logging data flow (#6272) Previously, tstun.Wrapper and magicsock.Conn managed their own statistics data structure and relied on an external call to Extract to extract (and reset) the statistics. This makes it difficult to ensure a maximum size on the statistics as the caller has no introspection into whether the number of unique connections is getting too large. Invert the control flow such that a *connstats.Statistics is registered with tstun.Wrapper and magicsock.Conn. Methods on non-nil *connstats.Statistics are called for every packet. This allows the implementation of connstats.Statistics (in the future) to better control when it needs to flush to ensure bounds on maximum sizes. The value registered into tstun.Wrapper and magicsock.Conn could be an interface, but that has two performance detriments: 1. Method calls on interface values are more expensive since they must go through a virtual method dispatch. 2. The implementation would need a sync.Mutex to protect the statistics value instead of using an atomic.Pointer. Given that methods on constats.Statistics are called for every packet, we want reduce the CPU cost on this hot path. Signed-off-by: Joe Tsai --- cmd/tailscaled/depaware.txt | 4 +- net/connstats/stats.go | 109 ++++++++++++++++++++++ net/{tunstats => connstats}/stats_test.go | 20 ++-- net/tstun/wrap.go | 32 ++----- net/tstun/wrap_test.go | 16 ++-- net/tunstats/stats.go | 70 -------------- wgengine/magicsock/magicsock.go | 67 +++---------- wgengine/magicsock/magicsock_test.go | 8 +- wgengine/netlog/logger.go | 32 +++---- wgengine/netlog/logger_test.go | 66 ------------- 10 files changed, 172 insertions(+), 252 deletions(-) create mode 100644 net/connstats/stats.go rename net/{tunstats => connstats}/stats_test.go (94%) delete mode 100644 net/tunstats/stats.go delete mode 100644 wgengine/netlog/logger_test.go diff --git a/cmd/tailscaled/depaware.txt b/cmd/tailscaled/depaware.txt index ee6af9952..33d82db5b 100644 --- a/cmd/tailscaled/depaware.txt +++ b/cmd/tailscaled/depaware.txt @@ -218,6 +218,7 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de tailscale.com/logtail/backoff from tailscale.com/control/controlclient+ tailscale.com/logtail/filch from tailscale.com/logpolicy tailscale.com/metrics from tailscale.com/derp+ + tailscale.com/net/connstats from tailscale.com/net/tstun+ tailscale.com/net/dns from tailscale.com/ipn/ipnlocal+ tailscale.com/net/dns/publicdns from tailscale.com/net/dns/resolver+ tailscale.com/net/dns/resolvconffile from tailscale.com/net/dns+ @@ -245,7 +246,6 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de tailscale.com/net/tsdial from tailscale.com/control/controlclient+ 💣 tailscale.com/net/tshttpproxy from tailscale.com/control/controlclient+ tailscale.com/net/tstun from tailscale.com/net/dns+ - tailscale.com/net/tunstats from tailscale.com/net/tstun tailscale.com/net/wsconn from tailscale.com/control/controlhttp+ tailscale.com/paths from tailscale.com/ipn/ipnlocal+ 💣 tailscale.com/portlist from tailscale.com/ipn/ipnlocal @@ -268,7 +268,7 @@ tailscale.com/cmd/tailscaled dependencies: (generated by github.com/tailscale/de tailscale.com/types/key from tailscale.com/control/controlbase+ tailscale.com/types/logger from tailscale.com/control/controlclient+ tailscale.com/types/logid from tailscale.com/logtail+ - tailscale.com/types/netlogtype from tailscale.com/net/tstun+ + tailscale.com/types/netlogtype from tailscale.com/net/connstats+ tailscale.com/types/netmap from tailscale.com/control/controlclient+ tailscale.com/types/nettype from tailscale.com/wgengine/magicsock+ tailscale.com/types/opt from tailscale.com/control/controlclient+ diff --git a/net/connstats/stats.go b/net/connstats/stats.go new file mode 100644 index 000000000..edd4707c4 --- /dev/null +++ b/net/connstats/stats.go @@ -0,0 +1,109 @@ +// Copyright (c) 2022 Tailscale Inc & AUTHORS All rights reserved. +// Use of this source code is governed by a BSD-style +// license that can be found in the LICENSE file. + +// Package connstats maintains statistics about connections +// flowing through a TUN device (which operate at the IP layer). +package connstats + +import ( + "net/netip" + "sync" + + "tailscale.com/net/packet" + "tailscale.com/types/netlogtype" +) + +// Statistics maintains counters for every connection. +// All methods are safe for concurrent use. +// The zero value is ready for use. +type Statistics struct { + mu sync.Mutex + virtual map[netlogtype.Connection]netlogtype.Counts + physical map[netlogtype.Connection]netlogtype.Counts +} + +// UpdateTxVirtual updates the counters for a transmitted IP packet +// The source and destination of the packet directly correspond with +// the source and destination in netlogtype.Connection. +func (s *Statistics) UpdateTxVirtual(b []byte) { + s.updateVirtual(b, false) +} + +// UpdateRxVirtual updates the counters for a received IP packet. +// The source and destination of the packet are inverted with respect to +// the source and destination in netlogtype.Connection. +func (s *Statistics) UpdateRxVirtual(b []byte) { + s.updateVirtual(b, true) +} + +func (s *Statistics) updateVirtual(b []byte, receive bool) { + var p packet.Parsed + p.Decode(b) + conn := netlogtype.Connection{Proto: p.IPProto, Src: p.Src, Dst: p.Dst} + if receive { + conn.Src, conn.Dst = conn.Dst, conn.Src + } + + s.mu.Lock() + defer s.mu.Unlock() + if s.virtual == nil { + s.virtual = make(map[netlogtype.Connection]netlogtype.Counts) + } + cnts := s.virtual[conn] + if receive { + cnts.RxPackets++ + cnts.RxBytes += uint64(len(b)) + } else { + cnts.TxPackets++ + cnts.TxBytes += uint64(len(b)) + } + s.virtual[conn] = cnts +} + +// UpdateTxPhysical updates the counters for a transmitted wireguard packet +// The src is always a Tailscale IP address, representing some remote peer. +// The dst is a remote IP address and port that corresponds +// with some physical peer backing the Tailscale IP address. +func (s *Statistics) UpdateTxPhysical(src netip.Addr, dst netip.AddrPort, n int) { + s.updatePhysical(src, dst, n, false) +} + +// UpdateRxPhysical updates the counters for a received wireguard packet. +// The src is always a Tailscale IP address, representing some remote peer. +// The dst is a remote IP address and port that corresponds +// with some physical peer backing the Tailscale IP address. +func (s *Statistics) UpdateRxPhysical(src netip.Addr, dst netip.AddrPort, n int) { + s.updatePhysical(src, dst, n, true) +} + +func (s *Statistics) updatePhysical(src netip.Addr, dst netip.AddrPort, n int, receive bool) { + conn := netlogtype.Connection{Src: netip.AddrPortFrom(src, 0), Dst: dst} + + s.mu.Lock() + defer s.mu.Unlock() + if s.physical == nil { + s.physical = make(map[netlogtype.Connection]netlogtype.Counts) + } + cnts := s.physical[conn] + if receive { + cnts.RxPackets++ + cnts.RxBytes += uint64(n) + } else { + cnts.TxPackets++ + cnts.TxBytes += uint64(n) + } + s.physical[conn] = cnts +} + +// Extract extracts and resets the counters for all active connections. +// It must be called periodically otherwise the memory used is unbounded. +func (s *Statistics) Extract() (virtual, physical map[netlogtype.Connection]netlogtype.Counts) { + s.mu.Lock() + defer s.mu.Unlock() + virtual = s.virtual + s.virtual = make(map[netlogtype.Connection]netlogtype.Counts) + physical = s.physical + s.physical = make(map[netlogtype.Connection]netlogtype.Counts) + return virtual, physical +} diff --git a/net/tunstats/stats_test.go b/net/connstats/stats_test.go similarity index 94% rename from net/tunstats/stats_test.go rename to net/connstats/stats_test.go index 1e73ce441..7c212e8aa 100644 --- a/net/tunstats/stats_test.go +++ b/net/connstats/stats_test.go @@ -2,7 +2,7 @@ // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. -package tunstats +package connstats import ( "encoding/binary" @@ -82,13 +82,13 @@ func TestConcurrent(t *testing.T) { cnts := gots[i][t2] if receive { - stats.UpdateRx(p) + stats.UpdateRxVirtual(p) cnts.RxPackets++ cnts.RxBytes += uint64(len(p)) } else { cnts.TxPackets++ cnts.TxBytes += uint64(len(p)) - stats.UpdateTx(p) + stats.UpdateTxVirtual(p) } gots[i][t2] = cnts time.Sleep(time.Duration(rn.Intn(1 + delay))) @@ -96,11 +96,13 @@ func TestConcurrent(t *testing.T) { }(i) } for range gots { - wants = append(wants, stats.Extract()) + virtual, _ := stats.Extract() + wants = append(wants, virtual) time.Sleep(time.Millisecond) } group.Wait() - wants = append(wants, stats.Extract()) + virtual, _ := stats.Extract() + wants = append(wants, virtual) got := make(map[netlogtype.Connection]netlogtype.Counts) want := make(map[netlogtype.Connection]netlogtype.Counts) @@ -126,7 +128,7 @@ func Benchmark(b *testing.B) { for i := 0; i < b.N; i++ { var s Statistics for j := 0; j < 1e3; j++ { - s.UpdateTx(p) + s.UpdateTxVirtual(p) } } }) @@ -138,7 +140,7 @@ func Benchmark(b *testing.B) { var s Statistics for j := 0; j < 1e3; j++ { binary.BigEndian.PutUint32(p[20:], uint32(j)) // unique port combination - s.UpdateTx(p) + s.UpdateTxVirtual(p) } } }) @@ -154,7 +156,7 @@ func Benchmark(b *testing.B) { go func() { defer group.Done() for k := 0; k < 1e3; k++ { - s.UpdateTx(p) + s.UpdateTxVirtual(p) } }() } @@ -179,7 +181,7 @@ func Benchmark(b *testing.B) { j *= 1e3 for k := 0; k < 1e3; k++ { binary.BigEndian.PutUint32(p[20:], uint32(j+k)) // unique port combination - s.UpdateTx(p) + s.UpdateTxVirtual(p) } }(j) } diff --git a/net/tstun/wrap.go b/net/tstun/wrap.go index a447e0ce6..64ee83065 100644 --- a/net/tstun/wrap.go +++ b/net/tstun/wrap.go @@ -22,15 +22,14 @@ "golang.zx2c4.com/wireguard/tun" "gvisor.dev/gvisor/pkg/tcpip/stack" "tailscale.com/disco" + "tailscale.com/net/connstats" "tailscale.com/net/packet" "tailscale.com/net/tsaddr" - "tailscale.com/net/tunstats" "tailscale.com/syncs" "tailscale.com/tstime/mono" "tailscale.com/types/ipproto" "tailscale.com/types/key" "tailscale.com/types/logger" - "tailscale.com/types/netlogtype" "tailscale.com/util/clientmetric" "tailscale.com/wgengine/filter" ) @@ -170,10 +169,7 @@ type Wrapper struct { disableTSMPRejected bool // stats maintains per-connection counters. - stats struct { - enabled atomic.Bool - tunstats.Statistics - } + stats atomic.Pointer[connstats.Statistics] } // tunReadResult is the result of a TUN read, or an injected result pretending to be a TUN read. @@ -568,8 +564,8 @@ func (t *Wrapper) Read(buf []byte, offset int) (int, error) { } } - if t.stats.enabled.Load() { - t.stats.UpdateTx(buf[offset:][:n]) + if stats := t.stats.Load(); stats != nil { + stats.UpdateTxVirtual(buf[offset:][:n]) } t.noteActivity() return n, nil @@ -701,8 +697,8 @@ func (t *Wrapper) Write(buf []byte, offset int) (int, error) { } func (t *Wrapper) tdevWrite(buf []byte, offset int) (int, error) { - if t.stats.enabled.Load() { - t.stats.UpdateRx(buf[offset:]) + if stats := t.stats.Load(); stats != nil { + stats.UpdateRxVirtual(buf[offset:]) } if t.isTAP { return t.tapWrite(buf, offset) @@ -843,18 +839,10 @@ func (t *Wrapper) Unwrap() tun.Device { return t.tdev } -// SetStatisticsEnabled enables per-connections packet counters. -// Disabling statistics gathering does not reset the counters. -// ExtractStatistics must be called to reset the counters and -// be periodically called while enabled to avoid unbounded memory use. -func (t *Wrapper) SetStatisticsEnabled(enable bool) { - t.stats.enabled.Store(enable) -} - -// ExtractStatistics extracts and resets the counters for all active connections. -// It must be called periodically otherwise the memory used is unbounded. -func (t *Wrapper) ExtractStatistics() map[netlogtype.Connection]netlogtype.Counts { - return t.stats.Extract() +// SetStatistics specifies a per-connection statistics aggregator. +// Nil may be specified to disable statistics gathering. +func (t *Wrapper) SetStatistics(stats *connstats.Statistics) { + t.stats.Store(stats) } var ( diff --git a/net/tstun/wrap_test.go b/net/tstun/wrap_test.go index 18acf3ba1..1a6b33cff 100644 --- a/net/tstun/wrap_test.go +++ b/net/tstun/wrap_test.go @@ -19,6 +19,7 @@ "go4.org/netipx" "golang.zx2c4.com/wireguard/tun/tuntest" "tailscale.com/disco" + "tailscale.com/net/connstats" "tailscale.com/net/netaddr" "tailscale.com/net/packet" "tailscale.com/tstest" @@ -283,11 +284,6 @@ func TestWriteAndInject(t *testing.T) { t.Errorf("%s not received", packet) } } - - // Statistics gathering is disabled by default. - if stats := tun.ExtractStatistics(); len(stats) > 0 { - t.Errorf("tun.ExtractStatistics = %v, want {}", stats) - } } func TestFilter(t *testing.T) { @@ -336,15 +332,17 @@ func TestFilter(t *testing.T) { }() var buf [MaxPacketSize]byte - tun.SetStatisticsEnabled(true) + stats := new(connstats.Statistics) + tun.SetStatistics(stats) for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { var n int var err error var filtered bool - if stats := tun.ExtractStatistics(); len(stats) > 0 { - t.Errorf("tun.ExtractStatistics = %v, want {}", stats) + tunStats, _ := stats.Extract() + if len(tunStats) > 0 { + t.Errorf("connstats.Statistics.Extract = %v, want {}", stats) } if tt.dir == in { @@ -377,7 +375,7 @@ func TestFilter(t *testing.T) { } } - got := tun.ExtractStatistics() + got, _ := stats.Extract() want := map[netlogtype.Connection]netlogtype.Counts{} if !tt.drop { var p packet.Parsed diff --git a/net/tunstats/stats.go b/net/tunstats/stats.go deleted file mode 100644 index 597ca9f6e..000000000 --- a/net/tunstats/stats.go +++ /dev/null @@ -1,70 +0,0 @@ -// Copyright (c) 2022 Tailscale Inc & AUTHORS All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -// Package tunstats maintains statistics about connections -// flowing through a TUN device (which operate at the IP layer). -package tunstats - -import ( - "sync" - - "tailscale.com/net/packet" - "tailscale.com/types/netlogtype" -) - -// Statistics maintains counters for every connection. -// All methods are safe for concurrent use. -// The zero value is ready for use. -type Statistics struct { - mu sync.Mutex - m map[netlogtype.Connection]netlogtype.Counts -} - -// UpdateTx updates the counters for a transmitted IP packet -// The source and destination of the packet directly correspond with -// the source and destination in netlogtype.Connection. -func (s *Statistics) UpdateTx(b []byte) { - s.update(b, false) -} - -// UpdateRx updates the counters for a received IP packet. -// The source and destination of the packet are inverted with respect to -// the source and destination in netlogtype.Connection. -func (s *Statistics) UpdateRx(b []byte) { - s.update(b, true) -} - -func (s *Statistics) update(b []byte, receive bool) { - var p packet.Parsed - p.Decode(b) - conn := netlogtype.Connection{Proto: p.IPProto, Src: p.Src, Dst: p.Dst} - if receive { - conn.Src, conn.Dst = conn.Dst, conn.Src - } - - s.mu.Lock() - defer s.mu.Unlock() - if s.m == nil { - s.m = make(map[netlogtype.Connection]netlogtype.Counts) - } - cnts := s.m[conn] - if receive { - cnts.RxPackets++ - cnts.RxBytes += uint64(len(b)) - } else { - cnts.TxPackets++ - cnts.TxBytes += uint64(len(b)) - } - s.m[conn] = cnts -} - -// Extract extracts and resets the counters for all active connections. -// It must be called periodically otherwise the memory used is unbounded. -func (s *Statistics) Extract() map[netlogtype.Connection]netlogtype.Counts { - s.mu.Lock() - defer s.mu.Unlock() - m := s.m - s.m = make(map[netlogtype.Connection]netlogtype.Counts) - return m -} diff --git a/wgengine/magicsock/magicsock.go b/wgengine/magicsock/magicsock.go index f3bfa8f73..6912b83d3 100644 --- a/wgengine/magicsock/magicsock.go +++ b/wgengine/magicsock/magicsock.go @@ -37,6 +37,7 @@ "tailscale.com/health" "tailscale.com/ipn/ipnstate" "tailscale.com/logtail/backoff" + "tailscale.com/net/connstats" "tailscale.com/net/dnscache" "tailscale.com/net/interfaces" "tailscale.com/net/netaddr" @@ -52,7 +53,6 @@ "tailscale.com/tstime/mono" "tailscale.com/types/key" "tailscale.com/types/logger" - "tailscale.com/types/netlogtype" "tailscale.com/types/netmap" "tailscale.com/types/nettype" "tailscale.com/util/clientmetric" @@ -337,19 +337,7 @@ type Conn struct { port atomic.Uint32 // stats maintains per-connection counters. - // See SetStatisticsEnabled and ExtractStatistics for details. - stats struct { - enabled atomic.Bool - - // TODO(joetsai): A per-Conn map of connections is easiest to implement. - // Since every packet occurs within the context of an endpoint, - // we could track the counts within the endpoint itself, - // and then merge the results when ExtractStatistics is called. - // That would avoid a map lookup for every packet. - - mu sync.Mutex - m map[netlogtype.Connection]netlogtype.Counts - } + stats atomic.Pointer[connstats.Statistics] // ============================================================ // mu guards all following fields; see userspaceEngine lock @@ -1754,8 +1742,8 @@ func (c *Conn) receiveIP(b []byte, ipp netip.AddrPort, cache *ippEndpointCache, ep = de } ep.noteRecvActivity() - if c.stats.enabled.Load() { - c.updateStats(ep.nodeAddr, ipp, netlogtype.Counts{RxPackets: 1, RxBytes: uint64(len(b))}) + if stats := c.stats.Load(); stats != nil { + stats.UpdateRxPhysical(ep.nodeAddr, ipp, len(b)) } return ep, true } @@ -1812,8 +1800,8 @@ func (c *Conn) processDERPReadResult(dm derpReadResult, b []byte) (n int, ep *en } ep.noteRecvActivity() - if c.stats.enabled.Load() { - c.updateStats(ep.nodeAddr, ipp, netlogtype.Counts{RxPackets: 1, RxBytes: uint64(dm.n)}) + if stats := c.stats.Load(); stats != nil { + stats.UpdateRxPhysical(ep.nodeAddr, ipp, dm.n) } return n, ep } @@ -3306,37 +3294,10 @@ func (c *Conn) UpdateStatus(sb *ipnstate.StatusBuilder) { }) } -// updateStats updates the statistics counters with the src, dst, and cnts. -// It is the caller's responsibility to check whether logging is enabled. -func (c *Conn) updateStats(src netip.Addr, dst netip.AddrPort, cnts netlogtype.Counts) { - conn := netlogtype.Connection{Src: netip.AddrPortFrom(src, 0), Dst: dst} - c.stats.mu.Lock() - defer c.stats.mu.Unlock() - mak.Set(&c.stats.m, conn, c.stats.m[conn].Add(cnts)) -} - -// SetStatisticsEnabled enables per-connection packet counters. -// Disabling statistics gathering does not reset the counters. -// ExtractStatistics must be called to reset the counters and -// be periodically called while enabled to avoid unbounded memory use. -func (c *Conn) SetStatisticsEnabled(enable bool) { - c.stats.enabled.Store(enable) -} - -// ExtractStatistics extracts and resets the counters for all active connections. -// It must be called periodically otherwise the memory used is unbounded. -// -// The source is always a peer's tailscale IP address, -// while the destination is the peer's physical IP address and port. -// As a special case, packets routed through DERP use a destination address -// of 127.3.3.40 with the port being the DERP region. -// This node's tailscale IP address never appears in the returned map. -func (c *Conn) ExtractStatistics() map[netlogtype.Connection]netlogtype.Counts { - c.stats.mu.Lock() - defer c.stats.mu.Unlock() - m := c.stats.m - c.stats.m = nil - return m +// SetStatistics specifies a per-connection statistics aggregator. +// Nil may be specified to disable statistics gathering. +func (c *Conn) SetStatistics(stats *connstats.Statistics) { + c.stats.Store(stats) } func ippDebugString(ua netip.AddrPort) string { @@ -3701,14 +3662,14 @@ func (de *endpoint) send(b []byte) error { var err error if udpAddr.IsValid() { _, err = de.c.sendAddr(udpAddr, de.publicKey, b) - if err == nil && de.c.stats.enabled.Load() { - de.c.updateStats(de.nodeAddr, udpAddr, netlogtype.Counts{TxPackets: 1, TxBytes: uint64(len(b))}) + if stats := de.c.stats.Load(); err == nil && stats != nil { + stats.UpdateTxPhysical(de.nodeAddr, udpAddr, len(b)) } } if derpAddr.IsValid() { if ok, _ := de.c.sendAddr(derpAddr, de.publicKey, b); ok { - if de.c.stats.enabled.Load() { - de.c.updateStats(de.nodeAddr, derpAddr, netlogtype.Counts{TxPackets: 1, TxBytes: uint64(len(b))}) + if stats := de.c.stats.Load(); stats != nil { + stats.UpdateTxPhysical(de.nodeAddr, derpAddr, len(b)) } if err != nil { // UDP failed but DERP worked, so good enough: diff --git a/wgengine/magicsock/magicsock_test.go b/wgengine/magicsock/magicsock_test.go index 15c9acfa2..c71ac32ab 100644 --- a/wgengine/magicsock/magicsock_test.go +++ b/wgengine/magicsock/magicsock_test.go @@ -35,6 +35,7 @@ "tailscale.com/derp/derphttp" "tailscale.com/disco" "tailscale.com/ipn/ipnstate" + "tailscale.com/net/connstats" "tailscale.com/net/netaddr" "tailscale.com/net/stun/stuntest" "tailscale.com/net/tstun" @@ -133,6 +134,7 @@ func runDERPAndStun(t *testing.T, logf logger.Logf, l nettype.PacketListener, st type magicStack struct { privateKey key.NodePrivate epCh chan []tailcfg.Endpoint // endpoint updates produced by this peer + stats connstats.Statistics // per-connection statistics conn *Conn // the magicsock itself tun *tuntest.ChannelTUN // TUN device to send/receive packets tsTun *tstun.Wrapper // wrapped tun that implements filtering and wgengine hooks @@ -1047,11 +1049,11 @@ func testTwoDevicePing(t *testing.T, d *devices) { } } - m1.conn.SetStatisticsEnabled(true) - m2.conn.SetStatisticsEnabled(true) + m1.conn.SetStatistics(&m1.stats) + m2.conn.SetStatistics(&m2.stats) checkStats := func(t *testing.T, m *magicStack, wantConns []netlogtype.Connection) { - stats := m.conn.ExtractStatistics() + _, stats := m.stats.Extract() for _, conn := range wantConns { if _, ok := stats[conn]; ok { return diff --git a/wgengine/netlog/logger.go b/wgengine/netlog/logger.go index 3d14da0f0..17d482b84 100644 --- a/wgengine/netlog/logger.go +++ b/wgengine/netlog/logger.go @@ -20,6 +20,7 @@ "golang.org/x/sync/errgroup" "tailscale.com/logpolicy" "tailscale.com/logtail" + "tailscale.com/net/connstats" "tailscale.com/net/tsaddr" "tailscale.com/smallzstd" "tailscale.com/tailcfg" @@ -34,14 +35,12 @@ // *tstun.Wrapper implements this interface. // *magicsock.Conn implements this interface. type Device interface { - SetStatisticsEnabled(bool) - ExtractStatistics() map[netlogtype.Connection]netlogtype.Counts + SetStatistics(*connstats.Statistics) } type noopDevice struct{} -func (noopDevice) SetStatisticsEnabled(bool) {} -func (noopDevice) ExtractStatistics() map[netlogtype.Connection]netlogtype.Counts { return nil } +func (noopDevice) SetStatistics(*connstats.Statistics) {} // Logger logs statistics about every connection. // At present, it only logs connections within a tailscale network. @@ -130,16 +129,15 @@ func (nl *Logger) Startup(nodeID tailcfg.StableNodeID, nodeLogID, domainLogID lo }, log.Printf) nl.logger = logger + stats := new(connstats.Statistics) ctx, cancel := context.WithCancel(context.Background()) nl.cancel = cancel nl.group.Go(func() error { - tun.SetStatisticsEnabled(true) - defer tun.SetStatisticsEnabled(false) - tun.ExtractStatistics() // clear out any stale statistics + tun.SetStatistics(stats) + defer tun.SetStatistics(nil) - sock.SetStatisticsEnabled(true) - defer sock.SetStatisticsEnabled(false) - sock.ExtractStatistics() // clear out any stale statistics + sock.SetStatistics(stats) + defer sock.SetStatistics(nil) start := time.Now() ticker := time.NewTicker(pollPeriod) @@ -147,22 +145,20 @@ func (nl *Logger) Startup(nodeID tailcfg.StableNodeID, nodeLogID, domainLogID lo var end time.Time select { case <-ctx.Done(): - tun.SetStatisticsEnabled(false) end = time.Now() case end = <-ticker.C: } - // NOTE: tunStats and sockStats will always be slightly out-of-sync. + // NOTE: connstats and sockStats will always be slightly out-of-sync. // It is impossible to have an atomic snapshot of statistics // at both layers without a global mutex that spans all layers. - tunStats := tun.ExtractStatistics() - sockStats := sock.ExtractStatistics() - if len(tunStats)+len(sockStats) > 0 { + connstats, sockStats := stats.Extract() + if len(connstats)+len(sockStats) > 0 { nl.mu.Lock() addrs := nl.addrs prefixes := nl.prefixes nl.mu.Unlock() - recordStatistics(logger, nodeID, start, end, tunStats, sockStats, addrs, prefixes) + recordStatistics(logger, nodeID, start, end, connstats, sockStats, addrs, prefixes) } if ctx.Err() != nil { @@ -175,7 +171,7 @@ func (nl *Logger) Startup(nodeID tailcfg.StableNodeID, nodeLogID, domainLogID lo return nil } -func recordStatistics(logger *logtail.Logger, nodeID tailcfg.StableNodeID, start, end time.Time, tunStats, sockStats map[netlogtype.Connection]netlogtype.Counts, addrs map[netip.Addr]bool, prefixes map[netip.Prefix]bool) { +func recordStatistics(logger *logtail.Logger, nodeID tailcfg.StableNodeID, start, end time.Time, connstats, sockStats map[netlogtype.Connection]netlogtype.Counts, addrs map[netip.Addr]bool, prefixes map[netip.Prefix]bool) { m := netlogtype.Message{NodeID: nodeID, Start: start.UTC(), End: end.UTC()} classifyAddr := func(a netip.Addr) (isTailscale, withinRoute bool) { @@ -194,7 +190,7 @@ func recordStatistics(logger *logtail.Logger, nodeID tailcfg.StableNodeID, start } exitTraffic := make(map[netlogtype.Connection]netlogtype.Counts) - for conn, cnts := range tunStats { + for conn, cnts := range connstats { srcIsTailscaleIP, srcWithinSubnet := classifyAddr(conn.Src.Addr()) dstIsTailscaleIP, dstWithinSubnet := classifyAddr(conn.Dst.Addr()) switch { diff --git a/wgengine/netlog/logger_test.go b/wgengine/netlog/logger_test.go deleted file mode 100644 index d7dd146f7..000000000 --- a/wgengine/netlog/logger_test.go +++ /dev/null @@ -1,66 +0,0 @@ -// Copyright (c) 2022 Tailscale Inc & AUTHORS All rights reserved. -// Use of this source code is governed by a BSD-style -// license that can be found in the LICENSE file. - -package netlog - -import ( - "context" - "net/http" - "testing" - - qt "github.com/frankban/quicktest" - "tailscale.com/logtail" - "tailscale.com/tstest" - "tailscale.com/types/netlogtype" - "tailscale.com/util/must" - "tailscale.com/wgengine/router" -) - -func init() { - testClient = &http.Client{Transport: &roundTripper} -} - -var roundTripper roundTripperFunc - -type roundTripperFunc struct { - F func(*http.Request) (*http.Response, error) -} - -func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { - return f.F(r) -} - -type fakeDevice struct { - toggled int // even => disabled, odd => enabled -} - -func (d *fakeDevice) SetStatisticsEnabled(enable bool) { - if enabled := d.toggled%2 == 1; enabled != enable { - d.toggled++ - } - -} -func (fakeDevice) ExtractStatistics() map[netlogtype.Connection]netlogtype.Counts { - // TODO(dsnet): Add a test that verifies that statistics are correctly - // extracted from the device and uploaded. Unfortunately, - // we can't reliably run this test until we fix http://go/oss/5856. - return nil -} - -func TestResourceCheck(t *testing.T) { - roundTripper.F = func(r *http.Request) (*http.Response, error) { - return &http.Response{StatusCode: 200}, nil - } - - c := qt.New(t) - tstest.ResourceCheck(t) - var l Logger - var d fakeDevice - for i := 0; i < 10; i++ { - must.Do(l.Startup("", logtail.PrivateID{}, logtail.PrivateID{}, &d, nil)) - l.ReconfigRoutes(&router.Config{}) - must.Do(l.Shutdown(context.Background())) - c.Assert(d.toggled, qt.Equals, 2*(i+1)) - } -}