diff --git a/examples/config.yml b/examples/config.yml index 1a6cb393..7b42bcc7 100644 --- a/examples/config.yml +++ b/examples/config.yml @@ -425,6 +425,13 @@ logging: # the route half of this setting to take effect. #ecn: true + # EXPERIMENTAL, Linux only. ecn_mark_threshold turns nebula into the AQM for its own receive queue — the one + # congested hop on a tunnel path that no kernel AQM can see. When the UDP receive queue's depth exceeds this + # fraction of the receive buffer (see listen.read_buffer), decapsulated ECT packets are CE-marked so ECN-capable + # senders back off before the queue overflows and regulates by tail-drop (loss) instead. 0 disables (default). + # Requires `ecn: true` end to end. Sampled once per receive batch. Reloadable. + #ecn_mark_threshold: 0.05 + # Nebula security group configuration firewall: # Action to take when a packet is not allowed by the firewall rules. diff --git a/outside.go b/outside.go index bbed1659..5f1b24a4 100644 --- a/outside.go +++ b/outside.go @@ -573,7 +573,15 @@ func (f *Interface) handleOutsideMessagePacket(hostinfo *HostInfo, out []byte, s // underlay into the inner header before firewall + TUN write. Other // outer codepoints are advisory only — we keep the inner unchanged. if f.ecnEnabled.Load() { - applyOuterECN(out, meta.OuterECN, hostinfo, f.l) + outerECN := meta.OuterECN + if meta.QueueCongested { + // nebula-as-AQM: our own receive queue is the congested hop on + // this path and no kernel AQM can see it. Depth beyond the + // marking threshold is treated as CE so ECT senders back off + // before the queue regulates by tail-drop instead. + outerECN = ecnCE + } + applyOuterECN(out, outerECN, hostinfo, f.l) } err := newPacket(out, true, fwPacket) diff --git a/pprof_debug.go b/pprof_debug.go index 5e1470d4..49ae5151 100644 --- a/pprof_debug.go +++ b/pprof_debug.go @@ -10,11 +10,13 @@ import ( _ "net/http/pprof" // registers pprof handlers on http.DefaultServeMux ) -// startPprofServer serves net/http/pprof on :6060 for the life of ctx. It is -// only compiled into debug builds (`-tags debug`, `make debug`), so a debug -// build announces itself with the Info line below. +// startPprofServer serves net/http/pprof on localhost:6060 for the life of +// ctx. It is only compiled into debug builds (`-tags debug`, `make debug`), +// so a debug build announces itself with the Info line below. Loopback only: +// a wildcard bind would expose profiles (peer addresses, config-derived +// state) to anything that can reach the host, the overlay included. func startPprofServer(ctx context.Context, l *slog.Logger) { - server := &http.Server{Addr: ":6060", Handler: nil} + server := &http.Server{Addr: "localhost:6060", Handler: nil} l.Info("Starting pprof debug server (debug build)", "addr", server.Addr) go func() { diff --git a/udp/conn.go b/udp/conn.go index 4af50b01..ab517b20 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -24,6 +24,13 @@ const MaxWriteBatch = 128 // supply on every packet. type RxMeta struct { OuterECN byte + // QueueCongested is set when the receiving socket's kernel queue depth + // exceeded the configured AQM marking threshold (tunnels.ecn_mark_threshold) + // when this batch was pulled. The decap path treats it like an outer CE + // mark on ECT inner packets — nebula acting as the AQM for the one queue + // on the tunnel path no kernel AQM can see. Backends without queue + // introspection leave it false. + QueueCongested bool } type EncReader func( diff --git a/udp/udp_linux.go b/udp/udp_linux.go index 356b8f88..0b00a2f2 100644 --- a/udp/udp_linux.go +++ b/udp/udp_linux.go @@ -8,8 +8,10 @@ import ( "errors" "fmt" "log/slog" + "math" "net" "net/netip" + "strconv" "sync/atomic" "syscall" "unsafe" @@ -43,6 +45,13 @@ type StdConn struct { // each arriving datagram as a per-slot cmsg, and ListenOut passes // the parsed value to the EncReader callback for RFC 6040 combine. ecnRecvSupported bool + + // ecnMarkThreshold holds tunnels.ecn_mark_threshold as float64 bits: the + // fraction of the socket receive buffer above which listenOutBatch flags + // the batch QueueCongested (decap then CE-marks ECT inner packets). Zero + // disables sampling entirely. Atomic because ReloadConfig may update it + // while the reader runs. + ecnMarkThreshold atomic.Uint64 } func NewListener(l *slog.Logger, ip netip.Addr, port int, multi bool, batch int) (Conn, error) { @@ -320,6 +329,24 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error { setMsgControllen(&msgs[i].Hdr, cmsgSpace) } } + + // AQM sample: one getsockopt per recvmmsg batch (skipped entirely at + // threshold 0). Sampled BEFORE the read: a single recvmmsg can drain + // more than the whole receive buffer (64 GRO superpackets ≈ 4MB), so + // post-read residue is ~always zero; the pre-read depth is the + // backlog that accumulated while the previous batch was processed — + // the actual standing-queue signal. Depth beyond the configured + // fraction of the receive buffer flags every packet in the batch so + // decap CE-marks ECT inner packets: the ECN substitute for the + // tail-drop this queue otherwise regulates with. + congested := false + if frac := math.Float64frombits(u.ecnMarkThreshold.Load()); frac > 0 { + var mi [unix.SK_MEMINFO_VARS]uint32 + if err := u.getMemInfo(&mi); err == nil { + congested = float64(mi[unix.SK_MEMINFO_RMEM_ALLOC]) >= frac*float64(mi[unix.SK_MEMINFO_RCVBUF]) + } + } + n, err := u.recvmmsg(msgs) if err != nil { if errors.Is(err, unix.EINTR) { @@ -340,7 +367,7 @@ func (u *StdConn) ListenOut(r EncReader, flush func()) error { segSize, outerECN = parseRecvCmsg(&msgs[i].Hdr, u.groSupported, u.ecnRecvSupported) } - deliverSegments(r, from, payload, segSize, RxMeta{OuterECN: outerECN}) + deliverSegments(r, from, payload, segSize, RxMeta{OuterECN: outerECN, QueueCongested: congested}) } flush() @@ -495,6 +522,8 @@ func writeSockaddr(buf []byte, addr netip.AddrPort, isV4 bool) (int, error) { } func (u *StdConn) ReloadConfig(c *config.C) { + u.reloadECNMarkThreshold(c) + b := c.GetInt("listen.read_buffer", 0) if b > 0 { if err := u.SetRecvBuffer(b); err == nil { @@ -536,6 +565,37 @@ func (u *StdConn) ReloadConfig(c *config.C) { } } +// reloadECNMarkThreshold parses tunnels.ecn_mark_threshold: the fraction +// (0..1] of the receive buffer above which decap CE-marks ECT inner packets. +// 0 (the default) disables the AQM sampling. Reloadable. +func (u *StdConn) reloadECNMarkThreshold(c *config.C) { + var frac float64 + switch v := c.Get("tunnels.ecn_mark_threshold").(type) { + case nil: + case float64: + frac = v + case int: + frac = float64(v) + case string: + f, err := strconv.ParseFloat(v, 64) + if err != nil { + u.l.Warn("tunnels.ecn_mark_threshold is not a number; disabling", "value", v) + } else { + frac = f + } + default: + u.l.Warn("tunnels.ecn_mark_threshold is not a number; disabling", "value", v) + } + if frac < 0 || frac > 1 { + u.l.Warn("tunnels.ecn_mark_threshold must be within [0, 1]; disabling", "value", frac) + frac = 0 + } + old := math.Float64frombits(u.ecnMarkThreshold.Swap(math.Float64bits(frac))) + if old != frac { + u.l.Info("tunnels.ecn_mark_threshold set", "fraction", frac) + } +} + func (u *StdConn) getMemInfo(meminfo *[unix.SK_MEMINFO_VARS]uint32) error { var vallen uint32 = 4 * unix.SK_MEMINFO_VARS _, _, err := unix.Syscall6(unix.SYS_GETSOCKOPT, uintptr(u.sysFd), uintptr(unix.SOL_SOCKET), uintptr(unix.SO_MEMINFO), uintptr(unsafe.Pointer(meminfo)), uintptr(unsafe.Pointer(&vallen)), 0)