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

feature/conn25: unify FlowTable storage to prepare for expiry

Browse Source 
Previously we had two maps keyed on a direction-specific tuple, with
distinct values containing the data (action) for that direction.
Values pointed at each other across maps to ensure they were removed
at the same time in the case of tuple overwrite, but LRU eviction
was per-map. So if LRU was turned on, it was possible for one
direction's data (action) to be evicted and leave the other direction
dangling.

NewFlow replaces the two direction-specific flow constructors, and
lookups return the direction-specific PacketAction directly.

Now the values in each map point to the same element, with data for both
directions in the element. A linked list also points to the elements to
implement LRU. The previous flowtrack.Cache is removed.

The single LRU structure will allow us to implement idle time expiration
by walking the list backward starting with the least recently used flow, and
stopping after a fixed number of flows, or at the first non-expired flow.

We add commented-out unused placeholder fields for tracking the
"last seen" timestamp, and an on-removal hook, to document the intent for
the follow-up expiry work.

Updates tailscale/corp#38630

Signed-off-by: Michael Ben-Ami <mzb@tailscale.com>
This commit is contained in:
Michael Ben-Ami
2026-05-26 10:09:48 -04:00
committed by mzbenami
parent 26952d53fa
commit 5877809097
3 changed files with 318 additions and 183 deletions
Download Patch File Download Diff File Expand all files Collapse all files
feature/conn25/flowtable.go
+90 -67
View File
@@ -4,7 +4,7 @@
package conn25
import (
"errors"
"container/list"
"sync"
"tailscale.com/net/flowtrack"
@@ -14,12 +14,23 @@ import (
// PacketAction may modify the packet.
type PacketAction func(*packet.Parsed)
// FlowData is an entry stored in the [FlowTable].
type FlowData struct {
// TupleAndAction wraps the [flowtrack.Tuple] and
// the [PacketAction] to return on lookups to that
// tuple.
type TupleAndAction struct {
Tuple flowtrack.Tuple
Action PacketAction
}
// FlowData is an entry stored in the [FlowTable]
// constructed by the consumer of the table.
// It specifies tuples and actions for each direction
// of the flow.
type FlowData struct {
FromTun TupleAndAction
FromWG TupleAndAction
}
// Origin is used to track the direction of a flow.
type Origin uint8
@@ -31,119 +42,131 @@ const (
FromWireGuard
)
// cachedFlow is the main unit of storage in the table.
// It wraps the [FlowData] passed in by the consumer, as well
// as internal metadata and callbacks.
type cachedFlow struct {
flow FlowData
paired flowtrack.Tuple // tuple for the other direction
data FlowData // user-defined tuples and actions for both directions
// lastSeen time.Time // tracks when the flow was last hit for expiration management
// onRemove func() // fires on removal/expiration (e.g. update watchers, send RST to client)
}
// FlowTable stores and retrieves [FlowData] that can be looked up
// by 5-tuple. New entries specify the tuple to use for both directions
// by 5-tuple [flowtrack.Tuple] and direction.
// New entries specify the tuple to use for both directions
// of traffic flow. The underlying cache is LRU, and the maximum number
// of entries is specified in calls to [NewFlowTable]. FlowTable has
// its own mutex and is safe for concurrent use.
type FlowTable struct {
mu sync.Mutex
fromTunCache *flowtrack.Cache[cachedFlow] // guarded by mu
fromWGCache *flowtrack.Cache[cachedFlow] // guarded by mu
fromTunCache map[flowtrack.Tuple]*list.Element
fromWGCache map[flowtrack.Tuple]*list.Element
lru *list.List
maxEntries int
}
// NewFlowTable returns a [FlowTable] maxEntries maximum entries.
// NewFlowTable returns a [FlowTable] with maxEntries maximum entries.
// A maxEntries of 0 indicates no maximum. See also [FlowTable].
func NewFlowTable(maxEntries int) *FlowTable {
return &FlowTable{
fromTunCache: &flowtrack.Cache[cachedFlow]{
MaxEntries: maxEntries,
},
fromWGCache: &flowtrack.Cache[cachedFlow]{
MaxEntries: maxEntries,
},
fromTunCache: make(map[flowtrack.Tuple]*list.Element, maxEntries),
fromWGCache: make(map[flowtrack.Tuple]*list.Element, maxEntries),
lru: list.New(),
maxEntries: maxEntries,
}
}
// LookupFromTunDevice looks up a [FlowData] entry that is valid to run for packets
// LookupFromTunDevice looks up a [PacketAction] that is valid to run on packets
// observed as coming from the tun device. The tuple must match the direction it was
// stored with.
func (t *FlowTable) LookupFromTunDevice(k flowtrack.Tuple) (FlowData, bool) {
func (t *FlowTable) LookupFromTunDevice(k flowtrack.Tuple) (PacketAction, bool) {
return t.lookup(k, FromTun)
}
// LookupFromWireGuard looks up a [FlowData] entry that is valid to run for packets
// LookupFromWireGuard looks up a [PacketAction] that is valid to run for packets
// observed as coming from the WireGuard tunnel. The tuple must match the direction it was
// stored with.
func (t *FlowTable) LookupFromWireGuard(k flowtrack.Tuple) (FlowData, bool) {
func (t *FlowTable) LookupFromWireGuard(k flowtrack.Tuple) (PacketAction, bool) {
return t.lookup(k, FromWireGuard)
}
func (t *FlowTable) lookup(k flowtrack.Tuple, want Origin) (FlowData, bool) {
var cache *flowtrack.Cache[cachedFlow]
switch want {
func (t *FlowTable) lookup(k flowtrack.Tuple, dir Origin) (PacketAction, bool) {
var cache map[flowtrack.Tuple]*list.Element
switch dir {
case FromTun:
cache = t.fromTunCache
case FromWireGuard:
cache = t.fromWGCache
default:
return FlowData{}, false
return nil, false
}
t.mu.Lock()
defer t.mu.Unlock()
v, ok := cache.Get(k)
ele, ok := cache[k]
if !ok {
return FlowData{}, false
}
return v.flow, true
}
// NewFlowFromTunDevice installs (or overwrites) both the forward and return entries.
// The forward tuple is tagged as FromTun, and the return tuple is tagged as FromWireGuard.
// If overwriting, it removes the old paired tuple for the forward key to avoid stale reverse mappings.
func (t *FlowTable) NewFlowFromTunDevice(fwd, rev FlowData) error {
return t.newFlow(FromTun, fwd, rev)
}
// NewFlowFromWireGuard installs (or overwrites) both the forward and return entries.
// The forward tuple is tagged as FromWireGuard, and the return tuple is tagged as FromTun.
// If overwriting, it removes the old paired tuple for the forward key to avoid stale reverse mappings.
func (t *FlowTable) NewFlowFromWireGuard(fwd, rev FlowData) error {
return t.newFlow(FromWireGuard, fwd, rev)
}
func (t *FlowTable) newFlow(fwdOrigin Origin, fwd, rev FlowData) error {
if fwd.Action == nil || rev.Action == nil {
return errors.New("nil action received for flow")
return nil, false
}
var fwdCache, revCache *flowtrack.Cache[cachedFlow]
switch fwdOrigin {
flow := ele.Value.(*cachedFlow)
var action PacketAction
switch dir {
case FromTun:
fwdCache, revCache = t.fromTunCache, t.fromWGCache
action = flow.data.FromTun.Action
case FromWireGuard:
fwdCache, revCache = t.fromWGCache, t.fromTunCache
default:
return errors.New("newFlow called with unknown direction")
action = flow.data.FromWG.Action
}
// Support LRU.
t.lru.MoveToFront(ele)
// TODO(mzb): Update flow.lastSeen.
return action, true
}
// NewFlow installs data as an flow in the table, and evicts any flow that
// either tuple already points at. This can result in two flows being evicted
// if each of the new tuples point at distinct existing flows. If the new flow
// would cause the table to exceed its maximum size, the least recently used
// (looked-up or created) flow is evicted. data is not validated, the caller must
// supply non-nil packet actions.
func (t *FlowTable) NewFlow(data FlowData) error {
t.mu.Lock()
defer t.mu.Unlock()
// If overwriting an existing entry, remove its previously-paired mapping so
// we don't leave stale tuples around.
if old, ok := fwdCache.Get(fwd.Tuple); ok {
revCache.Remove(old.paired)
}
if old, ok := revCache.Get(rev.Tuple); ok {
fwdCache.Remove(old.paired)
// If either tuple leads to anything existing, remove it.
t.removeFlowLocked(t.fromTunCache[data.FromTun.Tuple])
t.removeFlowLocked(t.fromWGCache[data.FromWG.Tuple])
flow := &cachedFlow{
data: data,
// Populate lastSeen
// Populate onRemove()
}
fwdCache.Add(fwd.Tuple, cachedFlow{
flow: fwd,
paired: rev.Tuple,
})
revCache.Add(rev.Tuple, cachedFlow{
flow: rev,
paired: fwd.Tuple,
})
ele := t.lru.PushFront(flow)
if t.maxEntries > 0 && t.lru.Len() > t.maxEntries {
t.removeFlowLocked(t.lru.Back())
}
t.fromTunCache[data.FromTun.Tuple] = ele
t.fromWGCache[data.FromWG.Tuple] = ele
return nil
}
func (t *FlowTable) removeFlowLocked(ele *list.Element) {
if ele == nil {
return
}
flow := t.lru.Remove(ele).(*cachedFlow)
delete(t.fromTunCache, flow.data.FromTun.Tuple)
delete(t.fromWGCache, flow.data.FromWG.Tuple)
// TODO(mzb): run flow.onRemove()
}