diff --git a/feature/routecheck/localapi.go b/feature/routecheck/localapi.go index 9b7282ec4..157674fdb 100644 --- a/feature/routecheck/localapi.go +++ b/feature/routecheck/localapi.go @@ -37,18 +37,16 @@ func serveRouteCheck(h *localapi.Handler, w http.ResponseWriter, r *http.Request return } - var err error - var report *routecheck.Report + report := rc.Report() if def.Bool(r.FormValue("probe"), false) { timeout := def.Duration(r.FormValue("timeout"), routecheck.DefaultTimeout) - timeout = min(max(0, timeout), 60*time.Second) // clamp to [0s, 60s] - report, err = rc.Refresh(r.Context(), timeout) - } else { - report = rc.Report() - } - if err != nil { - localapi.WriteErrorJSON(w, err) - return + timeout = clampRouteCheckTimeout(timeout) + rp, err := rc.Refresh(r.Context(), timeout) + if err != nil { + localapi.WriteErrorJSON(w, err) + return + } + report = rp } w.Header().Set("Content-Type", "application/json") @@ -61,3 +59,10 @@ func serveRouteCheck(h *localapi.Handler, w http.ResponseWriter, r *http.Request // with its default options, marshal with DefaultOptionsV1. jsonv2.MarshalWrite(w, report, jsonv1.DefaultOptionsV1()) } + +func clampRouteCheckTimeout(timeout time.Duration) time.Duration { + if timeout < 0 { + timeout = routecheck.DefaultTimeout + } + return min(max(0, timeout), 60*time.Second) // clamp to [0s, 60s] +} diff --git a/feature/routecheck/routecheck.go b/feature/routecheck/routecheck.go index d09bc09ca..5b0d6a92a 100644 --- a/feature/routecheck/routecheck.go +++ b/feature/routecheck/routecheck.go @@ -22,6 +22,7 @@ import ( "tailscale.com/net/routecheck" "tailscale.com/tailcfg" "tailscale.com/types/logger" + "tailscale.com/util/eventbus" ) // FeatureName is the name of the feature implemented by this package. @@ -43,6 +44,7 @@ type Extension struct { logf logger.Logf backend ipnext.SafeBackend + ec *eventbus.Client nb nodeBackender nm routecheck.NetMapper routers *RouterTracker @@ -83,7 +85,7 @@ func (e *Extension) Init(h ipnext.Host) error { pinger := e.backend.Sys().Engine.Get() - c, err := routecheck.NewClient(e.logf, e.nb, e.nm, pinger) + c, err := routecheck.NewClient(context.Background(), e.logf, e.nb, e.nm, pinger) if err != nil { return err } @@ -91,7 +93,11 @@ func (e *Extension) Init(h ipnext.Host) error { e.routers = TrackRouters(context.Background(), e.logf, ipnbus) e.routers.OnNetMapAvailable = e.Client.NotifyNetMapAvailable - e.routers.OnRoutersChange = e.incrementalRefresh + e.routers.OnRoutersChange = e.Client.NeedsIncrRefresh + + bus := e.backend.Sys().Bus.Get() + e.ec = bus.Client("routecheck") + eventbus.SubscribeFunc(e.ec, e.Client.WatchForNetMonRebind) // Watch for changes to the self node that would toggle the routecheck feature. e.reconcile.args = make(chan tailcfg.NodeView, 1) @@ -99,6 +105,9 @@ func (e *Extension) Init(h ipnext.Host) error { go e.reconcileLoop() h.Hooks().OnSelfChange.Add(e.reconcileWatcher) + // Probe for reachable peers. + go e.Client.Start() + return nil } @@ -109,22 +118,12 @@ func (e *Extension) Shutdown() error { close(e.reconcile.args) // lock prevents reconcileWatcher from writing to this channel e.reconcile.Unlock() + e.ec.Close() e.routers.Close() // stop the watcher before waiting for reconcile.done <-e.reconcile.done return e.Client.Close() } -func (e *Extension) needsRefresh() { - // TODO(sfllaw): Call e.Client.NeedsRefresh() after implementing it. -} - -func (e *Extension) incrementalRefresh(added, modified, removed []tailcfg.NodeID) { - // TODO(sfllaw): This refresh should be incremental, - // based on the added, modified, and removed nodes. - // Currently it refreshes everything. - e.needsRefresh() -} - // reconcileWatcher is called whenever e.routers should start, stop, or restart its watcher. // It may trigger a restart when self indicates that we have switched to a different tailnet or user, // in order to reset the internal state of e.routers and start tracking from scratch. @@ -159,7 +158,7 @@ func (e *Extension) reconcileLoop() { continue // can be started by toggling the nodeattr } if started { - e.needsRefresh() + e.Client.NeedsRefresh() } } } diff --git a/net/routecheck/report.go b/net/routecheck/report.go index 6f0d5e2c4..4d4f859cb 100644 --- a/net/routecheck/report.go +++ b/net/routecheck/report.go @@ -5,7 +5,6 @@ package routecheck import ( "cmp" - "context" "iter" "maps" "net/netip" @@ -29,17 +28,7 @@ var ( // It returns nil if a report isn’t available, which happens during initialization. func (c *Client) Report() *Report { metricReport.Add(1) - nm := c.nm.NetMapNoPeers() - if nm == nil { - return nil // The report wasn’t available. - } - - // TODO(sfllaw): Return the latest snapshot produced by background probing. - r, err := c.Refresh(context.TODO(), DefaultTimeout) - if err != nil { - c.logf("%v", err) - } - return r + return c.report.Load() } // Report contains the result of a single routecheck. diff --git a/net/routecheck/routecheck.go b/net/routecheck/routecheck.go index b8c44108e..5195b12b7 100644 --- a/net/routecheck/routecheck.go +++ b/net/routecheck/routecheck.go @@ -14,14 +14,17 @@ import ( "tailscale.com/envknob" "tailscale.com/ipn/ipnstate" + "tailscale.com/net/netmon" "tailscale.com/tailcfg" "tailscale.com/types/logger" "tailscale.com/types/netmap" "tailscale.com/util/clientmetric" + "tailscale.com/util/mak" ) var ( - metricRefresh = clientmetric.NewCounter("routecheck_refresh") + metricNeedsRefresh = clientmetric.NewCounter("routecheck_needs_refresh") + metricRefresh = clientmetric.NewCounter("routecheck_refresh") ) // DebugForceClientSideReachabilityRoutecheck reports whether routecheck should be forced on or off. @@ -58,6 +61,16 @@ type Client struct { nb NodeBackender nm NetMapper pinger Pinger + ctx context.Context + cancel context.CancelFunc + + // needsRefresh is sent a message by [Client.NeedsRefresh] + // to signal that a new report is needed. + // This message is received by the goroutine spawned by [Client.Start] + // which probes the appropriate routers to compile a new [Client.report]. + // This channel doesn’t need to be closed because the goroutine is canceled by ctx. + needsRefresh chan struct{} + report atomic.Pointer[Report] // needsRefresh signals that this needs refreshing // HasNetMap is a channel that can be closed to wake up goroutines // waiting for the netmap received after connecting to the control plane. @@ -116,7 +129,7 @@ type Pinger interface { } // NewClient returns a client that probes its peers using this LocalBackend. -func NewClient(logf logger.Logf, nb NodeBackender, nm NetMapper, pinger Pinger) (*Client, error) { +func NewClient(ctx context.Context, logf logger.Logf, nb NodeBackender, nm NetMapper, pinger Pinger) (*Client, error) { if nb == nil { return nil, errors.New("NodeBackender must be set") } @@ -126,11 +139,17 @@ func NewClient(logf logger.Logf, nb NodeBackender, nm NetMapper, pinger Pinger) if pinger == nil { return nil, errors.New("Pinger must be set") } + + ctx, cancel := context.WithCancel(ctx) c := &Client{ Logf: logf, nb: nb, nm: nm, pinger: pinger, + ctx: ctx, + cancel: cancel, + + needsRefresh: make(chan struct{}, 1), // debounce using buffer of 1 } c.hasNetMap.Store(new(make(chan struct{}))) return c, nil @@ -171,6 +190,8 @@ func (c *Client) waitForNetMap(ctx context.Context) (*netmap.NetworkMap, error) } select { + case <-c.ctx.Done(): // closed + return nil, c.ctx.Err() case <-ctx.Done(): return nil, ctx.Err() case <-*ch: // woken up by NotifyNetMapAvailable @@ -178,15 +199,150 @@ func (c *Client) waitForNetMap(ctx context.Context) (*netmap.NetworkMap, error) } } -// Refresh generates a new reachability report and returns it. +// Refresh generates and returns a new reachability report, after caching it in [Client.Report]. // A peer is considered unreachable if it doesn’t respond within the timeout. +// If the cached Client.Report is newer than the report that it just generated, +// Refresh will return the cached report instead of clobbering it report. func (c *Client) Refresh(ctx context.Context, timeout time.Duration) (*Report, error) { metricRefresh.Add(1) + c.vlogf("refreshing report") r, err := c.ProbeAllHARouters(ctx, 5, timeout) if err != nil { - return nil, fmt.Errorf("error probing routers: %w", err) + return nil, fmt.Errorf("error refreshing routers: %w", err) } - return r, nil + for { + saved := c.report.Load() + if saved != nil && !saved.Done.Before(r.Done) { + return saved, nil // don’t clobber newer reports + } + if c.report.CompareAndSwap(saved, r) { // retry if a concurrent Refresh stored first + c.vlogf("saved new report at %v", r.Done) + return r, nil + } + } +} + +// NeedsRefresh signals the need for a [Client.Refresh] to probe for a new report, +// which will be done in the background by [Client.Start]. +func (c *Client) NeedsRefresh() { + if !IsEnabled(c.nb.NodeBackend().Self()) { + return + } + + select { + case c.needsRefresh <- struct{}{}: + metricNeedsRefresh.Add(1) + c.vlogf("report needs refresh") + default: + // needsRefresh has already been raised, so debounce. + } +} + +// NeedsIncrRefresh signals the need for an incremental probe for a new report, +// because routers have been added, modified, or removed, +// which will be done in the background by [Client.Start]. +func (c *Client) NeedsIncrRefresh(added, modified, removed []tailcfg.NodeID) { + // TODO(sfllaw): Currently, this refreshes everything. + c.NeedsRefresh() +} + +// WatchForNetMonRebind watches the network monitor +// for a signal that the sockets need to be rebound, +// which implies that the cached report needs to be refreshed. +// See [netmon.ChangeDelta.RebindLikelyRequired]. +func (c *Client) WatchForNetMonRebind(delta netmon.ChangeDelta) { + if delta.RebindLikelyRequired { + c.NeedsRefresh() + } +} + +// Start runs periodic probes that compile routecheck reports. +// Use [Client.Close] to stop probing. +// Returns an error if the client has already been closed. +func (c *Client) Start() error { + if c.ctx.Err() != nil { + return c.ctx.Err() + } + + needsBootstrap := true + for { + select { + case <-c.needsRefresh: + nm := c.nm.NetMapWithPeers() + if nm == nil { + // There is no netmap: clear the cached report. + c.report.Store(nil) + needsBootstrap = true + continue + } + + if needsBootstrap { + r := c.bootstrap(nm) + c.report.Store(r) + needsBootstrap = false + } + + // TODO(sfllaw): Examine the shape of the overlapping + // routers and only probe if the routing table has + // changed sufficiently. For instance, a new router has + // come online or a router has been removed or a set of + // routers no longer overlap. + if _, err := c.Refresh(c.ctx, DefaultTimeout); err != nil { + c.logf("%v", err) + } + case <-c.ctx.Done(): // closed + return nil + } + } +} + +// bootstrap assumes that nodes that are connected to the control plane are reachable, +// while waiting for the first probe to finish. +// +// This function requires a netmap with peers. +func (c *Client) bootstrap(nm *netmap.NetworkMap) *Report { + if nm == nil { + return nil + } + + can4, can6 := supportsIPVersions(c.nb.NodeBackend().Self()) + if !can4 && !can6 { + return nil + } + addrFor := addrPicker(can4, can6) + + var r Report + for _, nodes := range GroupRoutersByPrefix(nm.Peers) { + if len(nodes) <= 1 { + continue // Not an overlapping router + } + + // TODO(sfllaw): Instead of trusting the Node.Online flag, + // which actually represents whether the node is connected + // to the control plane and not that it is reachable, + // we should cache reachability alongside the cached netmap + // long enough to survive a restart or a brief disconnection. + for _, n := range nodes { + if !n.Online().Get() { + continue // Not connected to the control plane. + } + + addr := addrFor(n) + if !addr.IsValid() { + continue // No valid addresses. + } + + mak.Set(&r.Reachable, n.ID(), Node{ + ID: n.ID(), + Name: n.Name(), + Addr: addr, + Routes: routes(n), + }) + } + } + r.Done = time.Now() + c.vlogf("bootstrapped report from netmap at %v", r.Done) + return &r } // Close immediately stops all active probes. @@ -195,6 +351,10 @@ func (c *Client) Close() error { return nil } + if c.cancel != nil { + c.cancel() + } + hasNetMap := c.hasNetMap.Swap(nil) // clear before waking anything up if hasNetMap != nil && *hasNetMap != nil { close(*hasNetMap) // wake waitForNetMap diff --git a/net/routecheck/routecheck_test.go b/net/routecheck/routecheck_test.go index 3984aea48..26148190e 100644 --- a/net/routecheck/routecheck_test.go +++ b/net/routecheck/routecheck_test.go @@ -145,7 +145,7 @@ func TestRefresh(t *testing.T) { withDelay(t, 10*time.Second)) } t.Cleanup(func() { b.Close() }) - c, err := routecheck.NewClient(t.Logf, b, b, b) + c, err := routecheck.NewClient(t.Context(), t.Logf, b, b, b) if err != nil { t.Fatalf("unexpected error: %v", err) } @@ -196,6 +196,64 @@ func TestRefresh(t *testing.T) { } } +func TestRefreshNewerReport(t *testing.T) { + cmpDiff := func(want, got any) string { + return gcmp.Diff(want, got, + gcmpopts.EquateComparable(netip.Addr{}, netip.Prefix{})) + } + peers := []tailcfg.NodeView{ + makeNode(11, withName("exit11"), withExitRoutes()), + makeNode(12, withName("exit12"), withExitRoutes()), + makeNode(21, withName("subnet21"), + withRoutes(netip.MustParsePrefix("192.168.1.0/24")), + withRoutes(netip.MustParsePrefix("2002:c000:0100::/48"))), + makeNode(22, withName("subnet22"), + withRoutes(netip.MustParsePrefix("192.168.1.0/24")), + withRoutes(netip.MustParsePrefix("2002:c000:0100::/48"))), + } + synctest.Test(t, func(t *testing.T) { + self := makeNode(99, withName("self")) + + // This is the “older” report where all the peers are online + // that we expect to get clobbered. + b := newStubBackend(self, peers) + t.Cleanup(func() { b.Close() }) + c, err := routecheck.NewClient(t.Context(), t.Logf, b, b, b) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + older, err := c.Refresh(t.Context(), routecheck.DefaultTimeout) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + // This is the “newer” report where some of the peers are online. + // Run this one before the actual test and fake its Done time. + time.Sleep(1 * time.Minute) + b.gone = set.Of(tailcfg.NodeID(11), tailcfg.NodeID(22)) + newer, err := c.Refresh(t.Context(), routecheck.DefaultTimeout) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !older.Done.Before(newer.Done) { + t.Errorf("newer report didn’t clobber older") + } + newer.Done = newer.Done.Add(1 * time.Hour) + + // Check that newer reports don’t get clobbered + // by simulating a run that happened between older and newer + // where only one node went offline. + b.gone = set.Of(tailcfg.NodeID(11)) + between, err := c.Refresh(t.Context(), routecheck.DefaultTimeout) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if diff := cmpDiff(between, newer); diff != "" { + t.Errorf("newer report was clobbered, -newer +between:\n%s", diff) + } + }) +} + func TestRoutersByPrefix(t *testing.T) { type routersByPrefix map[netip.Prefix][]tailcfg.NodeID simplify := func(rs routecheck.RoutersByPrefix) routersByPrefix { @@ -388,7 +446,7 @@ func TestRoutersByPrefix(t *testing.T) { self := makeNode(99, withName("self")) b := newStubBackend(self, tt.peers) t.Cleanup(func() { b.Close() }) - c, err := routecheck.NewClient(t.Logf, b, b, b) + c, err := routecheck.NewClient(t.Context(), t.Logf, b, b, b) if err != nil { t.Fatalf("unexpected error: %v", err) } diff --git a/net/routecheck/routes.go b/net/routecheck/routes.go index ad1b663e2..af9f4b5f3 100644 --- a/net/routecheck/routes.go +++ b/net/routecheck/routes.go @@ -24,8 +24,14 @@ type RoutersByPrefix map[netip.Prefix][]tailcfg.NodeView // RoutersByPrefix returns a map of nodes grouped by the subnet that they route. // See [RoutersByPrefix] for more detail. func (c *Client) RoutersByPrefix() RoutersByPrefix { + return GroupRoutersByPrefix(c.nb.NodeBackend().Peers()) +} + +// GroupRoutersByPrefix returns a map of nodes grouped by the subnet that they route. +// See [RoutersByPrefix] for more detail. +func GroupRoutersByPrefix(nodes []tailcfg.NodeView) RoutersByPrefix { var routers RoutersByPrefix - for _, n := range c.nb.NodeBackend().Peers() { + for _, n := range nodes { for _, pfx := range routes(n) { mak.Set(&routers, pfx, append(routers[pfx], n)) }