webnet/tailscale
1
0
Fork 0
Code Issues 1 Pull Requests Packages Projects Releases Wiki Activity

cmd/containerboot: track peers from IPN bus updates, stop using netmap.NetworkMap

Browse Source 
Some tests in another repo were broken by tailscale/tailscale#19607.
This fixes them, by finishing off the rest of the migration away from
netmap.NetworkMap on the IPN bus in containerboot.

Containerboot used to rebuild a full NetworkMap-shaped view while
reacting to IPN bus notifications. Now it insteads has its own
netmapState type (immutable) of exactly what it needs to track, and
sends those immutable values around, making cheap edits of new
immutable values when an IPN bus edit arrives.

This should make cmd/containerboot scale to much larger tailnets now too.

Fixes #19852
Fixes tailscale/corp#42347
Updates #12542

Change-Id: I88adaf061f85f677f954a764935e6654329d75a6
Signed-off-by: Brad Fitzpatrick <bradfitz@tailscale.com>
This commit is contained in:
Brad Fitzpatrick
2026-05-27 14:12:48 -07:00
committed by Brad Fitzpatrick
parent 80dc7a8d07
commit 364b952d62
15 changed files with 508 additions and 176 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/containerboot/main.go
+175 -51
View File
@@ -120,6 +120,7 @@ import (
"errors"
"fmt"
"io/fs"
"iter"
"log"
"math"
"net"
@@ -136,11 +137,12 @@ import (
"syscall"
"time"
"github.com/benbjohnson/immutable"
"golang.org/x/sys/unix"
"tailscale.com/client/local"
"tailscale.com/health"
"tailscale.com/ipn"
"tailscale.com/ipn/ipnstate"
kubeutils "tailscale.com/k8s-operator"
"tailscale.com/kube/authkey"
healthz "tailscale.com/kube/health"
@@ -151,7 +153,7 @@ import (
"tailscale.com/net/tsaddr"
"tailscale.com/tailcfg"
"tailscale.com/types/logger"
"tailscale.com/types/netmap"
"tailscale.com/types/views"
"tailscale.com/util/deephash"
"tailscale.com/util/dnsname"
"tailscale.com/util/linuxfw"
@@ -168,6 +170,137 @@ func getAutoAdvertiseBool() bool {
return defaultBool("TS_EXPERIMENTAL_SERVICE_AUTO_ADVERTISEMENT", true)
}
const containerbootWatchMask = ipn.NotifyInitialStatus |
ipn.NotifyPeerChanges |
ipn.NotifyNoNetMap
func notifyState(n ipn.Notify) (_ ipn.State, ok bool) {
if n.State != nil {
return *n.State, true
}
if n.InitialStatus != nil && n.InitialStatus.BackendState != "" {
if state, ok := ipn.StateFromString(n.InitialStatus.BackendState); ok {
return state, true
}
}
return ipn.NoState, false
}
var netmapStatePeerIDHasher = immutable.NewHasher(tailcfg.NodeID(0))
type netmapState struct {
self tailcfg.NodeView
peersByID *immutable.Map[tailcfg.NodeID, tailcfg.NodeView]
peersByName *immutable.Map[string, tailcfg.NodeView] // keyed by tailcfg.Node.Name when NodeID is unavailable
certDomains views.Slice[string]
dnsExtraRecords views.Slice[tailcfg.DNSRecord]
}
func (s netmapState) updateFromNotify(n ipn.Notify) netmapState {
if n.InitialStatus != nil {
s = s.updateFromStatus(n.InitialStatus)
}
if n.SelfChange != nil {
s.self = n.SelfChange.View()
}
for _, p := range n.PeersChanged {
s = s.upsertPeer(p.View())
}
for _, id := range n.PeersRemoved {
if s.peersByID != nil {
s.peersByID = s.peersByID.Delete(id)
}
}
return s
}
func (s netmapState) updateFromStatus(st *ipnstate.Status) netmapState {
s.certDomains = views.SliceOf(st.CertDomains)
s.dnsExtraRecords = views.SliceOf(st.ExtraRecords)
if st.Self != nil {
s.self = nodeFromPeerStatus(st.Self).View()
}
if len(st.Peer) != 0 {
s.peersByID = nil
s.peersByName = nil
for _, ps := range st.Peer {
s = s.upsertPeer(nodeFromPeerStatus(ps).View())
}
}
return s
}
func (s netmapState) upsertPeer(n tailcfg.NodeView) netmapState {
if !n.Valid() {
return s
}
if s.peersByID == nil {
s.peersByID = immutable.NewMap[tailcfg.NodeID, tailcfg.NodeView](netmapStatePeerIDHasher)
}
if s.peersByName == nil {
s.peersByName = immutable.NewMap[string, tailcfg.NodeView](nil)
}
if n.ID() != 0 {
s.peersByID = s.peersByID.Set(n.ID(), n)
if name := n.Name(); name != "" {
s.peersByName = s.peersByName.Delete(name)
}
return s
}
if n.Name() != "" {
s.peersByName = s.peersByName.Set(n.Name(), n)
}
return s
}
func nodeFromPeerStatus(ps *ipnstate.PeerStatus) *tailcfg.Node {
if ps == nil {
return nil
}
n := &tailcfg.Node{
ID: ps.NodeID,
StableID: ps.ID,
Name: ps.DNSName,
Key: ps.PublicKey,
}
for _, ip := range ps.TailscaleIPs {
n.Addresses = append(n.Addresses, netip.PrefixFrom(ip, ip.BitLen()))
}
if ps.AllowedIPs != nil {
n.AllowedIPs = ps.AllowedIPs.AsSlice()
}
return n
}
func (s netmapState) peers() iter.Seq[tailcfg.NodeView] {
return func(yield func(tailcfg.NodeView) bool) {
if s.peersByID != nil {
it := s.peersByID.Iterator()
for {
_, p, ok := it.Next()
if !ok {
break
}
if !yield(p) {
return
}
}
}
if s.peersByName != nil {
it := s.peersByName.Iterator()
for {
_, p, ok := it.Next()
if !ok {
break
}
if !yield(p) {
return
}
}
}
}
}
func main() {
if err := run(); err != nil && !errors.Is(err, context.Canceled) {
log.Fatal(err)
@@ -308,7 +441,7 @@ func run() error {
}
}
w, err := client.WatchIPNBus(bootCtx, ipn.NotifyInitialNetMap|ipn.NotifyInitialPrefs|ipn.NotifyInitialState|ipn.NotifyInitialHealthState|ipn.NotifyRateLimit)
w, err := client.WatchIPNBus(bootCtx, containerbootWatchMask|ipn.NotifyInitialPrefs|ipn.NotifyInitialHealthState)
if err != nil {
return fmt.Errorf("failed to watch tailscaled for updates: %w", err)
}
@@ -348,7 +481,7 @@ func run() error {
if err := tailscaleUp(bootCtx, cfg); err != nil {
return fmt.Errorf("failed to auth tailscale: %w", err)
}
w, err = client.WatchIPNBus(bootCtx, ipn.NotifyInitialNetMap|ipn.NotifyInitialState|ipn.NotifyRateLimit)
w, err = client.WatchIPNBus(bootCtx, containerbootWatchMask)
if err != nil {
return fmt.Errorf("rewatching tailscaled for updates after auth: %w", err)
}
@@ -368,8 +501,8 @@ authLoop:
return fmt.Errorf("failed to read from tailscaled: %w", err)
}
if n.State != nil {
switch *n.State {
if state, ok := notifyState(n); ok {
switch state {
case ipn.NeedsLogin:
if isOneStepConfig(cfg) {
// This could happen if this is the first time tailscaled was run for this
@@ -405,7 +538,7 @@ authLoop:
// deadline to continue monitoring for changes.
break authLoop
default:
log.Printf("tailscaled in state %q, waiting", *n.State)
log.Printf("tailscaled in state %q, waiting", state)
}
}
@@ -460,7 +593,7 @@ authLoop:
}
}
w, err = client.WatchIPNBus(ctx, ipn.NotifyInitialNetMap|ipn.NotifyInitialState|ipn.NotifyRateLimit)
w, err = client.WatchIPNBus(ctx, containerbootWatchMask)
if err != nil {
return fmt.Errorf("rewatching tailscaled for updates after auth: %w", err)
}
@@ -539,7 +672,7 @@ authLoop:
failedResolveAttempts++
}
var egressSvcsNotify chan *netmap.NetworkMap
var egressSvcsNotify chan netmapState
notifyChan := make(chan ipn.Notify)
errChan := make(chan error)
go func() {
@@ -553,12 +686,7 @@ authLoop:
}
}
}()
// Peer set changes (Add/Remove) no longer ride on the IPN bus; poll
// periodically so egress FQDN resolution and peer-aware work picks
// them up. SelfChange covers prompt self changes.
const peerPollInterval = 15 * time.Second
peerPoll := time.NewTicker(peerPollInterval)
defer peerPoll.Stop()
var nmState netmapState
var wg sync.WaitGroup
runLoop:
@@ -576,19 +704,17 @@ runLoop:
return fmt.Errorf("failed to read from tailscaled: %w", err)
case err := <-cfgWatchErrChan:
return fmt.Errorf("failed to watch tailscaled config: %w", err)
case <-peerPoll.C:
processNetmap = true
case n := <-notifyChan:
// TODO: (ChaosInTheCRD) Add node removed check when supported by ipn
if n.State != nil && *n.State != ipn.Running {
nmState = nmState.updateFromNotify(n)
if state, ok := notifyState(n); ok && state != ipn.Running {
// Something's gone wrong and we've left the authenticated state.
// Our container image never recovered gracefully from this, and the
// control flow required to make it work now is hard. So, just crash
// the container and rely on the container runtime to restart us,
// whereupon we'll go through initial auth again.
return fmt.Errorf("tailscaled left running state (now in state %q), exiting", *n.State)
return fmt.Errorf("tailscaled left running state (now in state %q), exiting", state)
}
if n.SelfChange != nil {
if n.InitialStatus != nil || n.SelfChange != nil || len(n.PeersChanged) != 0 || len(n.PeersRemoved) != 0 || len(n.PeerChangedPatch) != 0 {
processNetmap = true
}
case <-tc:
@@ -618,13 +744,12 @@ runLoop:
if !processNetmap {
continue
}
nm, err := fetchNetMap(ctx, client)
if err != nil {
log.Printf("error fetching netmap: %v", err)
self := nmState.self
if !self.Valid() {
continue
}
if nm != nil {
addrs = nm.SelfNode.Addresses().AsSlice()
{
addrs = self.Addresses().AsSlice()
newCurrentIPs := deephash.Hash(&addrs)
ipsHaveChanged := newCurrentIPs != currentIPs
@@ -636,14 +761,14 @@ runLoop:
// Kubernetes Secret to clean up tailnet nodes
// for proxies whose route setup continuously
// fails.
deviceID := nm.SelfNode.StableID()
deviceID := self.StableID()
if hasKubeStateStore(cfg) && deephash.Update(&currentDeviceID, &deviceID) {
if err := kc.storeDeviceID(ctx, nm.SelfNode.StableID()); err != nil {
if err := kc.storeDeviceID(ctx, deviceID); err != nil {
return fmt.Errorf("storing device ID in Kubernetes Secret: %w", err)
}
}
if cfg.TailnetTargetFQDN != "" {
egressAddrs, err := resolveTailnetFQDN(nm, cfg.TailnetTargetFQDN)
egressAddrs, err := resolveTailnetFQDN(nmState, cfg.TailnetTargetFQDN)
if err != nil {
log.Print(err.Error())
break
@@ -699,7 +824,10 @@ runLoop:
backendAddrs = newBackendAddrs
}
if cfg.ServeConfigPath != "" {
cd := certDomainFromNetmap(nm)
var cd string
if nmState.certDomains.Len() != 0 {
cd = nmState.certDomains.At(0)
}
if cd == "" {
cd = kubetypes.ValueNoHTTPS
}
@@ -742,9 +870,9 @@ runLoop:
// set up ensures that the operator does not
// advertize endpoints of broken proxies.
// TODO (irbekrm): instead of using the IP and FQDN, have some other mechanism for the proxy signal that it is 'Ready'.
deviceEndpoints := []any{nm.SelfNode.Name(), nm.SelfNode.Addresses()}
deviceEndpoints := []any{self.Name(), self.Addresses()}
if hasKubeStateStore(cfg) && deephash.Update(&currentDeviceEndpoints, &deviceEndpoints) {
if err := kc.storeDeviceEndpoints(ctx, nm.SelfNode.Name(), nm.SelfNode.Addresses().AsSlice()); err != nil {
if err := kc.storeDeviceEndpoints(ctx, self.Name(), addrs); err != nil {
return fmt.Errorf("storing device IPs and FQDN in Kubernetes Secret: %w", err)
}
}
@@ -773,7 +901,7 @@ runLoop:
}
if egressSvcsNotify != nil {
egressSvcsNotify <- nm
egressSvcsNotify <- nmState
}
}
if !startupTasksDone {
@@ -795,7 +923,7 @@ runLoop:
// will crash this node.
if cfg.EgressProxiesCfgPath != "" {
log.Printf("configuring egress proxy using configuration file at %s", cfg.EgressProxiesCfgPath)
egressSvcsNotify = make(chan *netmap.NetworkMap)
egressSvcsNotify = make(chan netmapState)
opts := egressProxyRunOpts{
cfgPath: cfg.EgressProxiesCfgPath,
nfr: nfr,
@@ -807,7 +935,7 @@ runLoop:
tailnetAddrs: addrs,
}
go func() {
if err := ep.run(ctx, nm, opts); err != nil {
if err := ep.run(ctx, nmState, opts); err != nil {
egressSvcsErrorChan <- err
}
}()
@@ -987,30 +1115,26 @@ func runHTTPServer(mux *http.ServeMux, addr string) (close func() error) {
}
}
// fetchNetMap fetches the current netmap from tailscaled via the
// "current-netmap" localapi debug action. The debug action's payload
// shape is intentionally not part of any stable API; containerboot
// reads its own internal-package types out of it. New external consumers
// should not rely on this — see [local.Client.Status] and friends.
func fetchNetMap(ctx context.Context, lc *local.Client) (*netmap.NetworkMap, error) {
return local.GetDebugResultJSON[*netmap.NetworkMap](ctx, lc, "current-netmap")
}
// resolveTailnetFQDN resolves a tailnet FQDN to a list of IP prefixes, which
// can be either a peer device, a Tailscale Service, or a 4via6 synthesized
// DNS name (e.g. "10-1-0-5-via-7.tailnet.ts.net").
func resolveTailnetFQDN(nm *netmap.NetworkMap, fqdn string) ([]netip.Prefix, error) {
func resolveTailnetFQDN(nm netmapState, fqdn string) ([]netip.Prefix, error) {
dnsFQDN, err := dnsname.ToFQDN(fqdn)
if err != nil {
return nil, fmt.Errorf("error parsing %q as FQDN: %w", fqdn, err)
}
// Check all peer devices first.
for _, p := range nm.Peers {
var ret []netip.Prefix
for p := range nm.peers() {
if strings.EqualFold(p.Name(), dnsFQDN.WithTrailingDot()) {
return p.Addresses().AsSlice(), nil
ret = p.Addresses().AsSlice()
break
}
}
if ret != nil {
return ret, nil
}
// If not found yet, check for a matching Tailscale Service.
if svcIPs := serviceIPsFromNetMap(nm, dnsFQDN); len(svcIPs) != 0 {
@@ -1020,7 +1144,7 @@ func resolveTailnetFQDN(nm *netmap.NetworkMap, fqdn string) ([]netip.Prefix, err
// If not found yet, check for a matching 4via6 DNS name.
if addr, ok := resolveViaDomain(dnsFQDN); ok {
prefix := netip.PrefixFrom(addr, addr.BitLen())
for _, nn := range nm.Peers {
for nn := range nm.peers() {
for _, allowedIP := range nn.AllowedIPs().All() {
if allowedIP.Contains(addr) {
return []netip.Prefix{prefix}, nil
@@ -1036,9 +1160,9 @@ func resolveTailnetFQDN(nm *netmap.NetworkMap, fqdn string) ([]netip.Prefix, err
// serviceIPsFromNetMap returns all IPs of a Tailscale Service if its FQDN is
// found in the netmap. Note that Tailscale Services are not a first-class
// object in the netmap, so we guess based on DNS ExtraRecords and AllowedIPs.
func serviceIPsFromNetMap(nm *netmap.NetworkMap, fqdn dnsname.FQDN) []netip.Prefix {
func serviceIPsFromNetMap(nm netmapState, fqdn dnsname.FQDN) []netip.Prefix {
var extraRecords []tailcfg.DNSRecord
for _, rec := range nm.DNS.ExtraRecords {
for _, rec := range nm.dnsExtraRecords.All() {
recFQDN, err := dnsname.ToFQDN(rec.Name)
if err != nil {
continue
@@ -1060,7 +1184,7 @@ func serviceIPsFromNetMap(nm *netmap.NetworkMap, fqdn dnsname.FQDN) []netip.Pref
continue
}
ipPrefix := netip.PrefixFrom(ip, ip.BitLen())
for _, ps := range nm.Peers {
for ps := range nm.peers() {
for _, allowedIP := range ps.AllowedIPs().All() {
if allowedIP == ipPrefix {
prefixes = append(prefixes, ipPrefix)