Something went wrong. Try again.
forked from tailscale/tailscale
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453// Copyright (c) Tailscale Inc & contributors// SPDX-License-Identifier: BSD-3-Clause
// Package netmapcache implements a persistent cache for [netmap.NetworkMap]// values, allowing a client to start up using stale but previously-valid state// even if a connection to the control plane is not immediately available.package netmapcache
import ( "cmp" "context" "crypto/sha256" "encoding/hex" jsonv1 "encoding/json" "errors" "fmt" "io/fs" "iter" "os" "path/filepath" "slices" "strings" "time"
"tailscale.com/feature/buildfeatures" "tailscale.com/tailcfg" "tailscale.com/tailcfg/nodecap" "tailscale.com/types/netmap" "tailscale.com/util/mak" "tailscale.com/util/set" "tailscale.com/wgengine/filter")
var ( // ErrKeyNotFound is a sentinel error reported by implementations of the [Store] // interface when loading a key that is not found in the store. ErrKeyNotFound = errors.New("storage key not found")
// ErrCacheNotAvailable is a sentinel error reported by cache methods when // the netmap caching feature is not enabled in the build. ErrCacheNotAvailable = errors.New("netmap cache is not available"))
// A Cache manages a columnar cache of a [netmap.NetworkMap]. Each Cache holds// a single netmap value; use [Cache.Store] to update or replace the cached// value and [Cache.Load] to read the cached value.type Cache struct { store Store
// wantKeys records the cache keys from the last write or load of a cached // netmap. This is used to prune keys that are no longer referenced after an // update. wantKeys set.Set[cacheKey]
// lastWrote records the last values written to each stored key. // // TODO(creachadair): This is meant to avoid disk writes, but I'm not // convinced we need it. Or maybe just track hashes of the content rather // than caching a complete copy. lastWrote map[cacheKey]lastWrote}
// NewCache constructs a new empty [Cache] from the given [Store].// It will panic if s == nil.func NewCache(s Store) *Cache { if s == nil { panic("a non-nil Store is required") } return &Cache{ store: s, wantKeys: make(set.Set[cacheKey]), lastWrote: make(map[cacheKey]lastWrote), }}
type lastWrote struct { digest string at time.Time}
func (c *Cache) writeJSON(ctx context.Context, key cacheKey, v any) error { j, err := jsonv1.Marshal(v) if err != nil { return fmt.Errorf("JSON marshalling %q: %w", key, err) }
// If the digest of this value has not changed since it was last written, // record that the key is of interest, but do not actually perform a write. last, ok := c.lastWrote[key] if ok && cacheDigest(j) == last.digest { c.wantKeys.Add(key) return nil }
if err := c.store.Store(ctx, string(key), j); err != nil { return err }
// Track the storage keys the current map is using, for storage GC. c.wantKeys.Add(key) c.lastWrote[key] = lastWrote{ digest: cacheDigest(j), at: time.Now(), } return nil}
func (c *Cache) removeUnwantedKeys(ctx context.Context) error { var errs []error for key, err := range c.store.List(ctx, "") { if err != nil { errs = append(errs, err) break } ckey := cacheKey(key) if !c.wantKeys.Contains(ckey) { if err := c.store.Remove(ctx, key); err != nil { errs = append(errs, fmt.Errorf("remove key %q: %w", key, err)) } delete(c.lastWrote, ckey) // even if removal failed, we don't want it } } return errors.Join(errs...)}
// FileStore implements the [Store] interface using a directory of files, in// which each key is encoded as a filename in the directory.// The caller is responsible to ensure the directory path exists before// using the store methods.type FileStore string
// List implements part of the [Store] interface.func (s FileStore) List(ctx context.Context, prefix string) iter.Seq2[string, error] { return func(yield func(string, error) bool) { des, err := os.ReadDir(string(s)) if os.IsNotExist(err) { return // nothing to read } else if err != nil { yield("", err) return }
// os.ReadDir reports entries already sorted, and the encoding preserves that. for _, de := range des { key, err := hex.DecodeString(de.Name()) if err != nil { yield("", err) return } name := string(key) if !strings.HasPrefix(name, prefix) { continue } else if !yield(name, nil) { return } } }}
// Load implements part of the [Store] interface.func (s FileStore) Load(ctx context.Context, key string) ([]byte, error) { data, err := os.ReadFile(filepath.Join(string(s), hex.EncodeToString([]byte(key)))) if errors.Is(err, os.ErrNotExist) { return nil, fmt.Errorf("key %q not found: %w", key, ErrKeyNotFound) } return data, err}
// Store implements part of the [Store] interface.func (s FileStore) Store(ctx context.Context, key string, value []byte) error { return os.WriteFile(filepath.Join(string(s), hex.EncodeToString([]byte(key))), value, 0600)}
// Remove implements part of the [Store] interface.func (s FileStore) Remove(ctx context.Context, key string) error { err := os.Remove(filepath.Join(string(s), hex.EncodeToString([]byte(key)))) if errors.Is(err, fs.ErrNotExist) { return nil } return err}
// cacheKey is a type wrapper for strings used as cache keys.type cacheKey string
const ( selfKey cacheKey = "self" miscKey cacheKey = "msic" dnsKey cacheKey = "dns" derpMapKey cacheKey = "derpmap" peerKeyPrefix cacheKey = "peer-" // + stable ID userKeyPrefix cacheKey = "user-" // + profile ID sshPolicyKey cacheKey = "ssh" packetFilterKey cacheKey = "filter")
// UpdateSelfOnly updates nm in the cache, replacing any previously cached// values for nm.Self and other tailnet metadata for the current node, but// skipping the Peers and UserProfiles fields. Any existing peer and profile// data are left unmodified.func (c *Cache) UpdateSelfOnly(ctx context.Context, nm *netmap.NetworkMap) error { if !buildfeatures.HasCacheNetMap || nm == nil || nm.Cached { return nil } if selfID := nm.User(); selfID == 0 { return errors.New("no user in netmap") } // Since we are not modifying peers or users (and in particular, not // removing any), we do not need to do any garbage collection on the storage // keys. return c.updateSelfOnly(ctx, nm)}
// Store records nm in the cache, replacing any previously-cached values.func (c *Cache) Store(ctx context.Context, nm *netmap.NetworkMap) error { if !buildfeatures.HasCacheNetMap || nm == nil || nm.Cached { return nil } if selfID := nm.User(); selfID == 0 { return errors.New("no user in netmap") }
// Because we may modify which peers and user profiles are valid, we need to // keep track of which storage keys we actually touch during the store. // This is used by c.removeUnwantedKeys to clean up keys that are no longer // referenced after peers and/or profiles are removed from nm. clear(c.wantKeys) if err := c.updateSelfOnly(ctx, nm); err != nil { return err } for _, p := range nm.Peers { key := peerKeyPrefix + cacheKey(p.StableID()) if err := c.writeJSON(ctx, key, netmapNode{Node: &p}); err != nil { return err } } for uid, u := range nm.UserProfiles { key := fmt.Sprintf("%s%d", userKeyPrefix, uid) if err := c.writeJSON(ctx, cacheKey(key), netmapUserProfile{UserProfile: &u}); err != nil { return err } } return c.removeUnwantedKeys(ctx)}
// UpdatePeers updates the cache to add or replace ("upsert") any peers// specified in update, and discard any peers specified in remove. This does// not affect any previous cached values for the self node or user profiles.//// It is expected that update and remove will be disjoint, but in the event// they are not, remove takes precedence. If the cache does not already contain// at least a self node, UpdatePeers reports an error without changing anything.func (c *Cache) UpdatePeers(ctx context.Context, update []tailcfg.NodeView, remove []tailcfg.StableNodeID) error { if !buildfeatures.HasCacheNetMap { return nil } if !c.wantKeys.Contains(selfKey) { return errors.New("no netmap is cached to apply an update") }
// Attempt all updates and removals before reporting any errors. var errs []error for _, p := range update { // We may already have an entry for this peer, or it may be new, but // rather than do a lookup we can just replace the existing entry if it // exists. key := peerKeyPrefix + cacheKey(p.StableID()) if err := c.writeJSON(ctx, key, netmapNode{Node: &p}); err != nil { errs = append(errs, err) continue } c.wantKeys.Add(key) } for _, q := range remove { key := peerKeyPrefix + cacheKey(q) if err := c.store.Remove(ctx, string(key)); err != nil { errs = append(errs, err) } c.wantKeys.Delete(key) delete(c.lastWrote, key) } return errors.Join(errs...)}
// updateSelfOnly updates the "static" parts of the netmap in the cache. It is// shared between [Cache.UpdateSelfOnly] and [Cache.Store].func (c *Cache) updateSelfOnly(ctx context.Context, nm *netmap.NetworkMap) error { if err := c.writeJSON(ctx, miscKey, netmapMisc{ MachineKey: &nm.MachineKey, CollectServices: &nm.CollectServices, DisplayMessages: &nm.DisplayMessages, TKAEnabled: &nm.TKAEnabled, TKAHead: &nm.TKAHead, Domain: &nm.Domain, DomainAuditLogID: &nm.DomainAuditLogID, }); err != nil { return err } if err := c.writeJSON(ctx, dnsKey, netmapDNS{DNS: &nm.DNS}); err != nil { return err } if err := c.writeJSON(ctx, derpMapKey, netmapDERPMap{DERPMap: &nm.DERPMap}); err != nil { return err } if err := c.writeJSON(ctx, selfKey, netmapNode{Node: &nm.SelfNode}); err != nil { return err
// N.B. The NodeKey and AllCaps fields can be recovered from SelfNode on // load, and do not need to be stored separately. } if err := c.writeJSON(ctx, packetFilterKey, netmapPacketFilter{Rules: &nm.PacketFilterRules}); err != nil { return err } if buildfeatures.HasSSH && nm.SSHPolicy != nil { if err := c.writeJSON(ctx, sshPolicyKey, netmapSSH{SSHPolicy: &nm.SSHPolicy}); err != nil { return err } } return nil}
// Load loads the cached [netmap.NetworkMap] value stored in c, if one is available.// It reports [ErrCacheNotAvailable] if no cached data are available.// On success, the Cached field of the returned network map is true.func (c *Cache) Load(ctx context.Context) (*netmap.NetworkMap, error) { if !buildfeatures.HasCacheNetMap { return nil, ErrCacheNotAvailable }
nm := netmap.NetworkMap{Cached: true}
// At minimum, we require that the cache contain a "self" node, or the data // are not usable. if self, err := c.store.Load(ctx, string(selfKey)); errors.Is(err, ErrKeyNotFound) { return nil, ErrCacheNotAvailable } else if err := jsonv1.Unmarshal(self, &netmapNode{Node: &nm.SelfNode}); err != nil { return nil, err } c.wantKeys.Add(selfKey)
// If we successfully recovered a SelfNode, pull out its related fields. if s := nm.SelfNode; s.Valid() { nm.NodeKey = s.Key() nm.AllCaps = make(set.Set[nodecap.Cap]) for _, c := range s.Capabilities().All() { nm.AllCaps.Add(c) } for c := range s.CapMap().All() { nm.AllCaps.Add(c) } }
// Unmarshal the contents of each specified cache entry directly into the // fields of the output. See the comment in types.go for more detail.
if err := c.readJSON(ctx, miscKey, &netmapMisc{ MachineKey: &nm.MachineKey, CollectServices: &nm.CollectServices, DisplayMessages: &nm.DisplayMessages, TKAEnabled: &nm.TKAEnabled, TKAHead: &nm.TKAHead, Domain: &nm.Domain, DomainAuditLogID: &nm.DomainAuditLogID, }); err != nil { return nil, err }
if err := c.readJSON(ctx, dnsKey, &netmapDNS{DNS: &nm.DNS}); err != nil { return nil, err } if err := c.readJSON(ctx, derpMapKey, &netmapDERPMap{DERPMap: &nm.DERPMap}); err != nil { return nil, err }
for key, err := range c.store.List(ctx, string(peerKeyPrefix)) { if err != nil { return nil, err } var peer tailcfg.NodeView if err := c.readJSON(ctx, cacheKey(key), &netmapNode{Node: &peer}); err != nil { return nil, err } nm.Peers = append(nm.Peers, peer) } slices.SortFunc(nm.Peers, func(a, b tailcfg.NodeView) int { return cmp.Compare(a.ID(), b.ID()) }) for key, err := range c.store.List(ctx, string(userKeyPrefix)) { if err != nil { return nil, err } var up tailcfg.UserProfileView if err := c.readJSON(ctx, cacheKey(key), &netmapUserProfile{UserProfile: &up}); err != nil { return nil, err } mak.Set(&nm.UserProfiles, up.ID(), up) } if err := c.readJSON(ctx, sshPolicyKey, &netmapSSH{SSHPolicy: &nm.SSHPolicy}); err != nil { return nil, err } if err := c.readJSON(ctx, packetFilterKey, &netmapPacketFilter{Rules: &nm.PacketFilterRules}); err != nil { return nil, err } else if r := nm.PacketFilterRules; r.Len() != 0 { // Reconstitute packet match expressions from the filter rules, nm.PacketFilter, err = filter.MatchesFromFilterRules(r.AsSlice()) if err != nil { return nil, err } }
return &nm, nil}
func (c *Cache) readJSON(ctx context.Context, key cacheKey, value any) error { data, err := c.store.Load(ctx, string(key)) if errors.Is(err, ErrKeyNotFound) { return nil } else if err != nil { return err } if err := jsonv1.Unmarshal(data, value); err != nil { return err } c.wantKeys.Add(key) c.lastWrote[key] = lastWrote{digest: cacheDigest(data), at: time.Now()} return nil}
// Store is the interface to persistent key-value storage used by a [Cache].type Store interface { // List lists all the stored keys having the specified prefixes, in // lexicographic order. // // Each pair yielded by the iterator is either a valid storage key and a nil // error, or an empty key and a non-nil error. After reporting an error, the // iterator must immediately return. List(ctx context.Context, prefix string) iter.Seq2[string, error]
// Load fetches the contents of the specified key. // If the key is not found in the store, Load must report [ErrKeyNotFound]. Load(ctx context.Context, key string) ([]byte, error)
// Store marshals and stores the contents of the specified value under key. // If the key already exists, its contents are replaced. Store(ctx context.Context, key string, value []byte) error
// Remove removes the specified key from the store. If the key does not exist, // Remove reports success (nil). Remove(ctx context.Context, key string) error}
// cacheDigest computes a string digest of the specified data, for use in// detecting cache hits.func cacheDigest(data []byte) string { h := sha256.Sum256(data); return string(h[:]) }