From ab2e751d4d17bb4599bd455d5a2f14404b2de78d Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Mon, 21 Sep 2026 12:07:41 -0400 Subject: [PATCH] fix(gateway): stop the edge XDP attach from taking a whole bond down Attaching a native XDP program makes a driver reallocate its rings, and a driver that has to do that takes the link down while it does. The attach walked every resolved interface back to back, so both members of a bonded uplink went down inside the same second and the node lost its only path off the box. Two changes, both about blast radius. The attach now waits for each bond slave to rejoin its aggregate before touching the next one. Carrier alone is not enough to call a slave ready: on an LACP bond a port can hold carrier, sit in the aggregator and be synchronized while collecting and distributing are both clear, and in that state it moves no packets. The wait reads the slave's MII status and its LACP actor state together, and gives up rather than attaching to the surviving member when a slave does not come back. The kernel is now asked, through the netdev generic netlink family, which interfaces can take a native program at all, before any of them is touched. A driver with no ndo_bpf was previously discovered by attaching, which meant the healthy slaves attached ahead of it had already paid a link bounce by the time the failure surfaced, and the rollback cost them another. Support is only ever read as a refusal when the answer is definite; a kernel too old to serve the family, or a query that fails on its own terms, warns and proceeds as before. Neither guard makes a bond with link monitoring switched off safe. With miimon at 0 the bonding driver never polls carrier, so a slave that bounces is left in a failed state that only a monitor it does not run would clear, and no amount of care on this side brings it back. That is fixed in the deployment's bond configuration, not here. These guards bound the damage; they do not repair a bond that cannot notice a link returning. The sequencing is extracted behind an internal seam so the ordering can be driven in tests without a real program or a real NIC, and the existing root-gated test that builds a real kernel bond exercises the new wait. Fixes #583 Co-Authored-By: Claude Opus 5 (1M context) --- NOTICE | 63 ++++- go.mod | 4 + go.sum | 12 +- internal/plumbing/ebpf/edgeattach/attach.go | 41 ++- internal/plumbing/ebpf/edgeattach/gate.go | 203 ++++++++++++++ .../plumbing/ebpf/edgeattach/gate_test.go | 257 ++++++++++++++++++ .../ebpf/edgepreflight/xdpfeatures.go | 138 ++++++++++ .../ebpf/edgepreflight/xdpfeatures_test.go | 68 +++++ 8 files changed, 773 insertions(+), 13 deletions(-) create mode 100644 internal/plumbing/ebpf/edgeattach/gate.go create mode 100644 internal/plumbing/ebpf/edgeattach/gate_test.go create mode 100644 internal/plumbing/ebpf/edgepreflight/xdpfeatures.go create mode 100644 internal/plumbing/ebpf/edgepreflight/xdpfeatures_test.go diff --git a/NOTICE b/NOTICE index b36b01d5..b7077960 100644 --- a/NOTICE +++ b/NOTICE @@ -2516,6 +2516,25 @@ THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLI ``` +## github.com/mdlayher/genetlink + +* Name: github.com/mdlayher/genetlink +* Version: v1.4.0 +* License: MIT + +``` +# MIT License + +Copyright (C) 2016-2022 Matt Layher + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +``` + ## github.com/mdlayher/ndp * Name: github.com/mdlayher/ndp @@ -2546,6 +2565,44 @@ CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. ``` +## github.com/mdlayher/netlink + +* Name: github.com/mdlayher/netlink +* Version: v1.9.0 +* License: MIT + +``` +# MIT License + +Copyright (C) 2016-2026 Matt Layher + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +``` + +## github.com/mdlayher/socket + +* Name: github.com/mdlayher/socket +* Version: v0.6.0 +* License: MIT + +``` +# MIT License + +Copyright (C) 2021 Matt Layher + +Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the "Software"), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE. + +``` + ## github.com/modern-go/concurrent * Name: github.com/modern-go/concurrent @@ -5474,7 +5531,7 @@ SOFTWARE. ## go.datum.net/network/api/v1alpha1 * Name: go.datum.net/network/api/v1alpha1 -* Version: v0.0.0-20260825185725-392ac243999b +* Version: v0.1.0 * License: AGPL-3.0 ``` @@ -7046,7 +7103,7 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. ## golang.org/x/sys * Name: golang.org/x/sys -* Version: v0.47.0 +* Version: v0.48.0 * License: BSD-3-Clause ``` @@ -7083,7 +7140,7 @@ OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. ## golang.org/x/term * Name: golang.org/x/term -* Version: v0.45.0 +* Version: v0.46.0 * License: BSD-3-Clause ``` diff --git a/go.mod b/go.mod index 831cf4c1..9681b990 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,9 @@ require ( github.com/coreos/go-iptables v0.8.0 github.com/kenshaw/baseconv v0.1.1 github.com/lorenzosaino/go-sysctl v0.3.1 + github.com/mdlayher/genetlink v1.4.0 github.com/mdlayher/ndp v1.1.0 + github.com/mdlayher/netlink v1.9.0 github.com/osrg/gobgp/v4 v4.9.0 github.com/prometheus/client_golang v1.24.1 github.com/prometheus/client_model v0.6.3 @@ -48,12 +50,14 @@ require ( github.com/go-openapi/swag v0.23.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/google/gnostic-models v0.7.0 // indirect + github.com/google/go-cmp v0.7.0 // indirect github.com/google/uuid v1.6.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect github.com/k-sone/critbitgo v1.4.0 // indirect github.com/mailru/easyjson v0.7.7 // indirect + github.com/mdlayher/socket v0.6.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect diff --git a/go.sum b/go.sum index 8a255f05..11c7080b 100644 --- a/go.sum +++ b/go.sum @@ -79,8 +79,6 @@ github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2 github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= -github.com/josharian/native v1.1.0 h1:uuaP0hAbW7Y4l0ZRQ6C9zfb7Mg1mbFKry/xzDAfmtLA= -github.com/josharian/native v1.1.0/go.mod h1:7X/raswPFr05uY3HiLlYeyQntB6OO7E/d2Cu7qoaN2w= github.com/jsimonetti/rtnetlink/v2 v2.0.1 h1:xda7qaHDSVOsADNouv7ukSuicKZO7GgVUCXxpaIEIlM= github.com/jsimonetti/rtnetlink/v2 v2.0.1/go.mod h1:7MoNYNbb3UaDHtF8udiJo/RH6VsTKP1pqKLUTVCvToE= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= @@ -106,12 +104,14 @@ github.com/lorenzosaino/go-sysctl v0.3.1 h1:3phX80tdITw2fJjZlwbXQnDWs4S30beNcMbw github.com/lorenzosaino/go-sysctl v0.3.1/go.mod h1:5grcsBRpspKknNS1qzt1eIeRDLrhpKZAtz8Fcuvs1Rc= github.com/mailru/easyjson v0.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= +github.com/mdlayher/genetlink v1.4.0 h1:f/Xs7Y2T+GyX9b3dbiUhnLE9InGs5F9RxJ2JwBMl71o= +github.com/mdlayher/genetlink v1.4.0/go.mod h1:d1hrKr8fwZU2JkcAtQUAzeTrI7nbgQSl+5k1cC0biSA= github.com/mdlayher/ndp v1.1.0 h1:QylGKGVtH60sKZUE88+IW5ila1Z/M9/OXhWdsVKuscs= github.com/mdlayher/ndp v1.1.0/go.mod h1:FmgESgemgjl38vuOIyAHWUUL6vQKA/pQNkvXdWsdQFM= -github.com/mdlayher/netlink v1.7.2 h1:/UtM3ofJap7Vl4QWCPDGXY8d3GIY2UGSDbK+QWmY8/g= -github.com/mdlayher/netlink v1.7.2/go.mod h1:xraEF7uJbxLhc5fpHL4cPe221LI2bdttWlU+ZGLfQSw= -github.com/mdlayher/socket v0.5.1 h1:VZaqt6RkGkt2OE9l3GcC6nZkqD3xKeQLyfleW/uBcos= -github.com/mdlayher/socket v0.5.1/go.mod h1:TjPLHI1UgwEv5J1B5q0zTZq12A/6H7nKmtTanQE37IQ= +github.com/mdlayher/netlink v1.9.0 h1:G8+GLq2x3v4D4MVIqDdNUhTUC7TKiCy/6MDkmItfKco= +github.com/mdlayher/netlink v1.9.0/go.mod h1:YBnl5BXsCoRuwBjKKlZ+aYmEoq0r12FDA/3JC+94KDg= +github.com/mdlayher/socket v0.6.0 h1:ScZPaAGyO1icQnbFrhPM8mnXyMu9qukC1K4ZoM2IQKU= +github.com/mdlayher/socket v0.6.0/go.mod h1:q7vozUAnxSqnjHc12Fik5yUKIzfZ8ITCfMkhOtE9z18= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= diff --git a/internal/plumbing/ebpf/edgeattach/attach.go b/internal/plumbing/ebpf/edgeattach/attach.go index 7cfa35ee..7794d8e0 100644 --- a/internal/plumbing/ebpf/edgeattach/attach.go +++ b/internal/plumbing/ebpf/edgeattach/attach.go @@ -153,6 +153,10 @@ func ResolveTargets(ifaceName string) ([]string, error) { // through ResolveTargets first, so this is usually a single interface and never // a bond master. // +// Every interface is checked for native XDP support before any of them is +// touched, and a bond slave is waited back into its aggregate before the next +// interface is attached, so a bonded uplink never loses every member at once. +// // If attaching one interface fails partway through, every link already attached // in this call is closed before returning, so a caller that gets an error holds // no partial attachment to clean up. @@ -164,20 +168,49 @@ func Attach(program *ebpf.Program, ifaceNames []string) ([]link.Link, error) { return nil, errors.New("edgeattach: no interfaces to attach to") } + return attachSequentially(ifaceNames, func(ifaceName string) (link.Link, error) { + return attachOne(program, ifaceName) + }) +} + +// attachSequentially is Attach's ordering, with the attach step itself passed +// in so tests can drive the sequence without a real program or a real NIC. +// +// Nothing is attached until every interface has been checked, and each bond +// slave is waited back into its aggregate before the next interface is +// touched. Both orderings are load-bearing on a bonded uplink. +func attachSequentially(ifaceNames []string, attach func(string) (link.Link, error)) ([]link.Link, error) { + if err := checkNativeXDPSupport(ifaceNames); err != nil { + return nil, err + } + links := make([]link.Link, 0, len(ifaceNames)) for _, ifaceName := range ifaceNames { - xdpLink, err := attachOne(program, ifaceName) + xdpLink, err := attach(ifaceName) if err != nil { - for _, already := range links { - _ = already.Close() - } + closeAttached(links) return nil, err } links = append(links, xdpLink) + + if err := waitBondSlaveReady(ifaceName, BondReadyTimeout); err != nil { + closeAttached(links) + return nil, err + } } return links, nil } +// closeAttached unwinds the links attached so far in one Attach call. +func closeAttached(links []link.Link) { + for _, already := range links { + if already == nil { + continue + } + _ = already.Close() + } +} + // attachOne attaches program to ifaceName's XDP hook in native driver mode, the // single-interface mechanism Attach applies across its list. func attachOne(program *ebpf.Program, ifaceName string) (link.Link, error) { diff --git a/internal/plumbing/ebpf/edgeattach/gate.go b/internal/plumbing/ebpf/edgeattach/gate.go new file mode 100644 index 00000000..3be443b5 --- /dev/null +++ b/internal/plumbing/ebpf/edgeattach/gate.go @@ -0,0 +1,203 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgeattach + +import ( + "errors" + "fmt" + "log/slog" + "os" + "path/filepath" + "time" + + "github.com/vishvananda/netlink" + + "go.datum.net/galactic/internal/plumbing/ebpf/edgepreflight" +) + +// Attaching a native XDP program makes a driver reallocate its rings, and a +// driver that has to do that takes the link down while it does. On 10GBASE-T +// copper the relink alone runs to several seconds. On an ordinary uplink that +// is a visible but survivable blip. On a bond slave it is neither, for two +// reasons this file exists to address. +// +// First, attaching to every slave at once takes every member of the aggregate +// down inside the same second, so the node loses its uplink outright rather +// than riding out each bounce on the surviving member. waitBondSlaveReady +// makes the attach wait for each slave to rejoin its aggregate before the next +// one is touched. +// +// Second, a driver that cannot take a native program at all is found out by +// attaching, which means the healthy slaves attached ahead of it have already +// paid a bounce by the time the failure surfaces, and the rollback costs them +// another. checkNativeXDPSupport asks the kernel which interfaces support the +// program before any link is touched, so an unsupported one fails the process +// with nothing disturbed. +// +// Neither guard makes a bond whose link monitoring is switched off safe. With +// miimon at 0 the bonding driver never polls carrier, so a slave that bounces +// is left in a failed state that only a monitor it does not run would clear. +// These guards bound the damage; they do not repair a bond that cannot detect +// a link coming back. + +// BondReadyTimeout bounds how long Attach waits for one bond slave to rejoin +// its aggregate before giving up on it. It covers a driver's ring realloc plus +// LACP re-convergence with the switch, so it is generous by design: the cost of +// waiting too long is a slow start, and the cost of waiting too little is +// attaching to the next slave while this one is still down. +var BondReadyTimeout = 45 * time.Second + +// bondReadyPollInterval is how often the aggregate is re-read while waiting. +const bondReadyPollInterval = 250 * time.Millisecond + +// LACP actor state bits, as carried in IFLA_BOND_SLAVE_AD_ACTOR_OPER_PORT_STATE +// and defined by IEEE 802.1AX. A port that is collecting and distributing is +// carrying traffic; one that is merely synchronized is not. +const ( + lacpStateCollecting = 1 << 4 + lacpStateDistributing = 1 << 5 +) + +// linkByIndexFn is an override point, matching linkByNameFn and linkListFn, so +// the bond-mode lookup can be faked in tests. +var linkByIndexFn = netlink.LinkByIndex + +// sleepFn is an override point so the readiness poll does not spend real time +// in tests. +var sleepFn = time.Sleep + +// checkNativeXDPSupport refuses the whole attach if the kernel positively +// reports that one of ifaceNames cannot run a native XDP program. +// +// It refuses only on a definite answer. A kernel too old to serve the netdev +// generic netlink family, or a query that fails for its own reasons, leaves +// support unknown, and an unknown is not grounds to keep a gateway down: those +// log and proceed, which is the behavior every release before this one had. +func checkNativeXDPSupport(ifaceNames []string) error { + for _, ifaceName := range ifaceNames { + supported, err := edgepreflight.InterfaceXDPFn(ifaceName) + switch { + case errors.Is(err, edgepreflight.ErrXDPFeaturesUnavailable): + slog.Warn("edgeattach: kernel cannot report per-interface XDP support, attaching without checking first "+ + "(a driver that cannot take the program will be found out by attaching, which bounces the link)", + "interface", ifaceName) + case err != nil: + slog.Warn("edgeattach: could not read XDP support for interface, attaching without checking first", + "interface", ifaceName, "err", err) + case !supported: + return fmt.Errorf( + "edgeattach: interface %q (driver %q) does not support native XDP, refusing to attach the edge "+ + "gateway datapath to any interface: attaching to the rest would leave traffic arriving on %q "+ + "bypassing the datapath entirely, and on a bonded uplink it would take the link down", + ifaceName, driverName(ifaceName), ifaceName) + } + } + return nil +} + +// driverName reports the kernel driver bound to ifaceName, for an error message +// naming the hardware an operator has to act on. Best-effort: an interface with +// no backing device, such as a veth, reports "unknown". +func driverName(ifaceName string) string { + target, err := os.Readlink(filepath.Join("/sys/class/net", ifaceName, "device", "driver")) + if err != nil { + return "unknown" + } + return filepath.Base(target) +} + +// waitBondSlaveReady blocks until ifaceName is carrying traffic in its bond +// again, or until BondReadyTimeout elapses. +// +// An interface that is not enslaved to a bond returns immediately: there is no +// aggregate to rejoin and no sibling link to protect. Note that the interface +// named in the gateway's configuration can itself be a slave rather than a +// master -- a deployment that points at one member of a bond directly still +// gets the wait, because the bounce still costs its siblings. +func waitBondSlaveReady(ifaceName string, timeout time.Duration) error { + slave, masterIndex, ok, err := bondSlaveOf(ifaceName) + if err != nil { + return err + } + if !ok { + return nil + } + + lacp, err := bondUsesLACP(masterIndex) + if err != nil { + return err + } + + deadline := time.Now().Add(timeout) + for { + if bondSlaveCarryingTraffic(slave, lacp) { + return nil + } + if time.Now().After(deadline) { + return fmt.Errorf( + "edgeattach: bond slave %q did not rejoin its aggregate within %s of the XDP attach "+ + "(mii status %d, LACP actor state %d); refusing to attach to any remaining slave, "+ + "because doing so would take the rest of the bond down with it", + ifaceName, timeout, slave.MiiStatus, slave.AdActorOperPortState) + } + sleepFn(bondReadyPollInterval) + + slave, _, ok, err = bondSlaveOf(ifaceName) + if err != nil { + return err + } + if !ok { + // It left the bond while we waited. There is no aggregate to + // wait for any more, and nothing here can put it back. + return nil + } + } +} + +// bondSlaveOf reports ifaceName's bond membership, if it has one, along with +// the ifindex of the master it is enslaved to. +func bondSlaveOf(ifaceName string) (slave *netlink.BondSlave, masterIndex int, ok bool, err error) { + l, err := linkByNameFn(ifaceName) + if err != nil { + return nil, 0, false, fmt.Errorf("edgeattach: find link %q: %w", ifaceName, err) + } + s, ok := l.Attrs().Slave.(*netlink.BondSlave) + if !ok { + return nil, 0, false, nil + } + return s, l.Attrs().MasterIndex, true, nil +} + +// bondUsesLACP reports whether the bonding master at masterIndex runs 802.3ad. +// Only an LACP bond has actor port state to read; every other mode carries +// nothing beyond the MII status. +func bondUsesLACP(masterIndex int) (bool, error) { + master, err := linkByIndexFn(masterIndex) + if err != nil { + return false, fmt.Errorf("edgeattach: find bonding master at index %d: %w", masterIndex, err) + } + b, ok := master.(*netlink.Bond) + if !ok { + return false, nil + } + return b.Mode == netlink.BOND_MODE_802_3AD, nil +} + +// bondSlaveCarryingTraffic reports whether slave is back to carrying traffic. +// +// On an LACP bond that means more than carrier: a slave can hold carrier, sit +// in the aggregator and be synchronized while collecting and distributing are +// both clear, which is exactly the state an XDP attach leaves behind on a bond +// with link monitoring switched off, and in that state it moves no packets. +func bondSlaveCarryingTraffic(slave *netlink.BondSlave, lacp bool) bool { + if slave.MiiStatus != netlink.BondLinkUp { + return false + } + if !lacp { + return true + } + const want = lacpStateCollecting | lacpStateDistributing + return slave.AdActorOperPortState&want == want +} diff --git a/internal/plumbing/ebpf/edgeattach/gate_test.go b/internal/plumbing/ebpf/edgeattach/gate_test.go new file mode 100644 index 00000000..e1730c95 --- /dev/null +++ b/internal/plumbing/ebpf/edgeattach/gate_test.go @@ -0,0 +1,257 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgeattach + +import ( + "errors" + "slices" + "strings" + "testing" + "time" + + "github.com/cilium/ebpf/link" + "github.com/vishvananda/netlink" + + "go.datum.net/galactic/internal/plumbing/ebpf/edgepreflight" +) + +// fakeSlaveLink is a netlink.Link carrying bond slave data, the view +// waitBondSlaveReady reads. +type fakeSlaveLink struct { + netlink.LinkAttrs +} + +func (f *fakeSlaveLink) Attrs() *netlink.LinkAttrs { return &f.LinkAttrs } +func (f *fakeSlaveLink) Type() string { return "veth" } + +// Interface names these tests drive: one slave whose driver takes a native +// program, one whose driver does not. +const ( + gateSlave = "enp3s0f1" + gateSlaveNoXDP = "eno2" + gateSlaveSib = "enp6s0f1" + gateMasterIdx = 7 +) + +// slaveLink builds a link enslaved to the test bond with the given LACP actor +// state and MII status. +func slaveLink(name string, mii netlink.BondSlaveMiiStatus, actorState uint8) netlink.Link { + return &fakeSlaveLink{LinkAttrs: netlink.LinkAttrs{ + Name: name, + MasterIndex: gateMasterIdx, + Slave: &netlink.BondSlave{ + MiiStatus: mii, + AdActorOperPortState: actorState, + }, + }} +} + +// lacpMaster builds the 802.3ad bonding master these tests enslave to. +func lacpMaster() netlink.Link { + b := netlink.NewLinkBond(netlink.LinkAttrs{Name: "bond0", Index: gateMasterIdx}) + b.Mode = netlink.BOND_MODE_802_3AD + return b +} + +const ( + // lacpCarrying is the actor state of a port that is collecting and + // distributing: activity, aggregation, sync, collecting, distributing. + lacpCarrying = 0x3d + // lacpSyncOnly is the state an XDP attach leaves behind on a bond whose + // link monitoring is off: synchronized, but neither collecting nor + // distributing. A port in this state moves no packets. + lacpSyncOnly = 0x0d +) + +// restoreGateVars puts every override this file swaps back after a test. +func restoreGateVars(t *testing.T) { + t.Helper() + byName, byIndex, sleep, xdp := linkByNameFn, linkByIndexFn, sleepFn, edgepreflight.InterfaceXDPFn + t.Cleanup(func() { + linkByNameFn, linkByIndexFn, sleepFn, edgepreflight.InterfaceXDPFn = byName, byIndex, sleep, xdp + }) +} + +func TestWaitBondSlaveReady_NonSlaveReturnsImmediately(t *testing.T) { + restoreGateVars(t) + + linkByNameFn = func(name string) (netlink.Link, error) { + return &fakeSlaveLink{LinkAttrs: netlink.LinkAttrs{Name: name}}, nil + } + sleepFn = func(time.Duration) { t.Fatal("a plain interface must not be polled") } + + if err := waitBondSlaveReady("eth0", time.Second); err != nil { + t.Fatalf("waitBondSlaveReady on a non-slave: %v", err) + } +} + +func TestWaitBondSlaveReady_WaitsForCollectingAndDistributing(t *testing.T) { + restoreGateVars(t) + + // The slave reports sync-only twice before it starts carrying traffic, + // the sequence a real slave walks after its driver finishes the ring + // realloc and LACP re-converges with the switch. + states := []uint8{lacpSyncOnly, lacpSyncOnly, lacpCarrying} + var reads int + linkByNameFn = func(name string) (netlink.Link, error) { + state := states[min(reads, len(states)-1)] + reads++ + return slaveLink(name, netlink.BondLinkUp, state), nil + } + linkByIndexFn = func(int) (netlink.Link, error) { return lacpMaster(), nil } + + var slept int + sleepFn = func(time.Duration) { slept++ } + + if err := waitBondSlaveReady(gateSlave, time.Minute); err != nil { + t.Fatalf("waitBondSlaveReady: %v", err) + } + if slept != 2 { + t.Errorf("polled %d times, want 2 (one per sync-only read)", slept) + } +} + +func TestWaitBondSlaveReady_SyncedButNotCarryingTimesOut(t *testing.T) { + restoreGateVars(t) + + // Carrier is up and the port is synchronized, but collecting and + // distributing never arrive. That is the state a slave is left in when + // the bond cannot detect its link returning, and it is not readiness. + linkByNameFn = func(name string) (netlink.Link, error) { + return slaveLink(name, netlink.BondLinkUp, lacpSyncOnly), nil + } + linkByIndexFn = func(int) (netlink.Link, error) { return lacpMaster(), nil } + sleepFn = func(time.Duration) {} + + err := waitBondSlaveReady(gateSlave, time.Millisecond) + if err == nil { + t.Fatal("a slave stuck synchronized-but-not-carrying must not report ready") + } + if !strings.Contains(err.Error(), "did not rejoin its aggregate") { + t.Errorf("error does not name the failure: %v", err) + } +} + +func TestWaitBondSlaveReady_CarrierDownIsNotReady(t *testing.T) { + restoreGateVars(t) + + // The first thing an XDP attach does to a slave is take carrier away + // while the driver reallocates its rings. Until it comes back the slave + // is not ready, whatever its actor state says. + linkByNameFn = func(name string) (netlink.Link, error) { + return slaveLink(name, netlink.BondLinkDown, lacpCarrying), nil + } + linkByIndexFn = func(int) (netlink.Link, error) { return lacpMaster(), nil } + sleepFn = func(time.Duration) {} + + if err := waitBondSlaveReady(gateSlave, time.Millisecond); err == nil { + t.Fatal("a slave with no carrier must not report ready") + } +} + +func TestWaitBondSlaveReady_NonLACPBondNeedsOnlyCarrier(t *testing.T) { + restoreGateVars(t) + + // A balance-rr bond carries no actor state at all, so waiting for + // collecting and distributing there would never return. + linkByNameFn = func(name string) (netlink.Link, error) { + return slaveLink(name, netlink.BondLinkUp, 0), nil + } + linkByIndexFn = func(int) (netlink.Link, error) { + return netlink.NewLinkBond(netlink.LinkAttrs{Name: "bond0", Index: gateMasterIdx}), nil + } + sleepFn = func(time.Duration) { t.Fatal("a carrier-up non-LACP slave must not be polled") } + + if err := waitBondSlaveReady(gateSlave, time.Second); err != nil { + t.Fatalf("waitBondSlaveReady on a non-LACP bond: %v", err) + } +} + +func TestCheckNativeXDPSupport_RefusesBeforeTouchingAnyLink(t *testing.T) { + restoreGateVars(t) + + var asked []string + edgepreflight.InterfaceXDPFn = func(name string) (bool, error) { + asked = append(asked, name) + return name != gateSlaveNoXDP, nil + } + + err := checkNativeXDPSupport([]string{gateSlave, gateSlaveNoXDP}) + if err == nil { + t.Fatal("an interface with no native XDP support must fail the whole attach") + } + if !strings.Contains(err.Error(), gateSlaveNoXDP) { + t.Errorf("error does not name the unsupported interface: %v", err) + } + if len(asked) != 2 { + t.Errorf("checked %v, want both interfaces checked", asked) + } +} + +func TestCheckNativeXDPSupport_UnknownSupportProceeds(t *testing.T) { + restoreGateVars(t) + + // A kernel too old to answer, and a query that fails on its own terms, + // both leave support unknown. Neither is grounds to hold the gateway + // down, so both proceed with a warning. + for name, probeErr := range map[string]error{ + "family unavailable": edgepreflight.ErrXDPFeaturesUnavailable, + "query failed": errors.New("netlink: connection reset"), + } { + t.Run(name, func(t *testing.T) { + edgepreflight.InterfaceXDPFn = func(string) (bool, error) { return false, probeErr } + if err := checkNativeXDPSupport([]string{gateSlave}); err != nil { + t.Fatalf("unknown support must not refuse the attach: %v", err) + } + }) + } +} + +func TestAttachSequentially_StopsAtTheFirstSlaveThatDoesNotComeBack(t *testing.T) { + restoreGateVars(t) + + // The regression this whole change exists for: with two slaves and the + // first one failing to rejoin, the second must never be attached. It is + // the surviving uplink, and taking it down is what puts the node dark. + edgepreflight.InterfaceXDPFn = func(string) (bool, error) { return true, nil } + linkByNameFn = func(name string) (netlink.Link, error) { + return slaveLink(name, netlink.BondLinkUp, lacpSyncOnly), nil + } + linkByIndexFn = func(int) (netlink.Link, error) { return lacpMaster(), nil } + sleepFn = func(time.Duration) {} + + BondReadyTimeout = time.Millisecond + t.Cleanup(func() { BondReadyTimeout = 45 * time.Second }) + + var attached []string + _, err := attachSequentially([]string{gateSlave, gateSlaveSib}, func(name string) (link.Link, error) { + attached = append(attached, name) + return nil, nil + }) + if err == nil { + t.Fatal("attach must fail when a slave does not rejoin its aggregate") + } + if want := []string{gateSlave}; !slices.Equal(attached, want) { + t.Errorf("attached %v, want %v -- the second slave must be left alone", attached, want) + } +} + +func TestAttachSequentially_AttachesNothingWhenSupportIsMissing(t *testing.T) { + restoreGateVars(t) + + edgepreflight.InterfaceXDPFn = func(name string) (bool, error) { return name != gateSlaveNoXDP, nil } + + var attached []string + _, err := attachSequentially([]string{gateSlave, gateSlaveNoXDP}, func(name string) (link.Link, error) { + attached = append(attached, name) + return nil, nil + }) + if err == nil { + t.Fatal("attach must fail when an interface cannot take a native program") + } + if len(attached) != 0 { + t.Errorf("attached %v, want nothing -- the check runs before any link is touched", attached) + } +} diff --git a/internal/plumbing/ebpf/edgepreflight/xdpfeatures.go b/internal/plumbing/ebpf/edgepreflight/xdpfeatures.go new file mode 100644 index 00000000..434636bc --- /dev/null +++ b/internal/plumbing/ebpf/edgepreflight/xdpfeatures.go @@ -0,0 +1,138 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgepreflight + +import ( + "errors" + "fmt" + "net" + "os" + + "github.com/mdlayher/genetlink" + "github.com/mdlayher/netlink" +) + +// The netdev generic netlink family reports, per interface, which XDP actions +// that interface's driver implements. It is the only way to learn whether a +// native XDP attach will be accepted without performing one: a driver with no +// ndo_bpf rejects the attach, but a driver that has one reallocates its rings +// to service it and drops carrier while it does. On a bond slave that carrier +// drop can cost the aggregate a member, so support has to be known before any +// link is touched. +// +// Constants are from include/uapi/linux/netdev.h. +const ( + netdevFamilyName = "netdev" + + netdevCmdDevGet = 1 + + netdevAttrDevIfindex = 1 + netdevAttrDevXDPFeatures = 3 + netdevAttrDevXDPZCMaxSegs = 4 + + // netdevXDPActBasic is the feature bit for the XDP actions every + // XDP-capable driver implements: XDP_ABORTED, XDP_DROP, XDP_PASS and + // XDP_TX. The edge datapath needs no more than these, so it is the + // whole of what this checks. + netdevXDPActBasic = 1 +) + +// ErrXDPFeaturesUnavailable reports that the running kernel does not serve the +// netdev generic netlink family, so per-interface XDP support cannot be read. +// The family landed in Linux 6.3; on anything older a caller has no choice but +// to discover support by attaching. +var ErrXDPFeaturesUnavailable = errors.New("edgepreflight: netdev generic netlink family unavailable") + +// ifaceByNameFn is an override point so the ifindex lookup can be faked in +// tests alongside the family query. +var ifaceByNameFn = net.InterfaceByName + +// InterfaceXDPFn is an override point, as elsewhere in this codebase, so tests +// can substitute a fake capability view without touching the host network +// stack. +var InterfaceXDPFn = InterfaceSupportsNativeXDP + +// InterfaceSupportsNativeXDP reports whether ifaceName's driver implements the +// basic XDP actions, meaning a native-mode attach against it will be accepted. +// +// It returns [ErrXDPFeaturesUnavailable] when the kernel does not serve the +// netdev family at all, which a caller should treat as "cannot tell" rather +// than "unsupported". +func InterfaceSupportsNativeXDP(ifaceName string) (bool, error) { + ifi, err := ifaceByNameFn(ifaceName) + if err != nil { + return false, fmt.Errorf("edgepreflight: find interface %q: %w", ifaceName, err) + } + + conn, err := genetlink.Dial(nil) + if err != nil { + return false, fmt.Errorf("edgepreflight: dial generic netlink: %w", err) + } + defer conn.Close() //nolint:errcheck // read-only socket, nothing to flush + + family, err := conn.GetFamily(netdevFamilyName) + if err != nil { + if errors.Is(err, os.ErrNotExist) { + return false, ErrXDPFeaturesUnavailable + } + return false, fmt.Errorf("edgepreflight: resolve the %q generic netlink family: %w", netdevFamilyName, err) + } + + enc := netlink.NewAttributeEncoder() + enc.Uint32(netdevAttrDevIfindex, uint32(ifi.Index)) //nolint:gosec // an ifindex is never negative + data, err := enc.Encode() + if err != nil { + return false, fmt.Errorf("edgepreflight: encode netdev request for %q: %w", ifaceName, err) + } + + msgs, err := conn.Execute( + genetlink.Message{ + Header: genetlink.Header{Command: netdevCmdDevGet, Version: family.Version}, + Data: data, + }, + family.ID, + netlink.Request, + ) + if err != nil { + return false, fmt.Errorf("edgepreflight: query XDP features of %q: %w", ifaceName, err) + } + + for _, msg := range msgs { + features, ok, err := xdpFeaturesFrom(msg.Data) + if err != nil { + return false, fmt.Errorf("edgepreflight: decode XDP features of %q: %w", ifaceName, err) + } + if ok { + return features&netdevXDPActBasic != 0, nil + } + } + // The interface exists and the family answered, but reported no feature + // attribute. Treat that as "cannot tell" rather than claiming the driver + // is unsupported on the strength of a missing field. + return false, ErrXDPFeaturesUnavailable +} + +// xdpFeaturesFrom pulls the XDP feature bitmask out of one netdev reply, +// reporting whether the attribute was present at all. +func xdpFeaturesFrom(data []byte) (uint64, bool, error) { + dec, err := netlink.NewAttributeDecoder(data) + if err != nil { + return 0, false, err + } + var ( + features uint64 + found bool + ) + for dec.Next() { + if dec.Type() == netdevAttrDevXDPFeatures { + features = dec.Uint64() + found = true + } + } + if err := dec.Err(); err != nil { + return 0, false, err + } + return features, found, nil +} diff --git a/internal/plumbing/ebpf/edgepreflight/xdpfeatures_test.go b/internal/plumbing/ebpf/edgepreflight/xdpfeatures_test.go new file mode 100644 index 00000000..9bdb95b1 --- /dev/null +++ b/internal/plumbing/ebpf/edgepreflight/xdpfeatures_test.go @@ -0,0 +1,68 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package edgepreflight + +import ( + "testing" + + "github.com/mdlayher/netlink" +) + +// encodeFeatures builds the attribute payload a netdev reply carries, so the +// decoder is exercised against the same wire shape the kernel produces. +func encodeFeatures(t *testing.T, attrType uint16, value uint64) []byte { + t.Helper() + enc := netlink.NewAttributeEncoder() + enc.Uint32(netdevAttrDevIfindex, 3) + enc.Uint64(attrType, value) + data, err := enc.Encode() + if err != nil { + t.Fatalf("encode netdev reply: %v", err) + } + return data +} + +func TestXDPFeaturesFrom(t *testing.T) { + tests := map[string]struct { + data []byte + wantFeatures uint64 + wantFound bool + }{ + "driver reports the basic actions": { + data: encodeFeatures(t, netdevAttrDevXDPFeatures, netdevXDPActBasic|2|4), + wantFeatures: netdevXDPActBasic | 2 | 4, + wantFound: true, + }, + // A driver with no ndo_bpf at all is reported with an empty + // feature set rather than by omitting the attribute. + "driver reports no actions": { + data: encodeFeatures(t, netdevAttrDevXDPFeatures, 0), + wantFeatures: 0, + wantFound: true, + }, + "reply carries no feature attribute": { + data: encodeFeatures(t, netdevAttrDevXDPZCMaxSegs, 1), + wantFound: false, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + features, found, err := xdpFeaturesFrom(tc.data) + if err != nil { + t.Fatalf("xdpFeaturesFrom: %v", err) + } + if found != tc.wantFound { + t.Fatalf("found = %v, want %v", found, tc.wantFound) + } + if features != tc.wantFeatures { + t.Errorf("features = %#x, want %#x", features, tc.wantFeatures) + } + if tc.wantFound && (features&netdevXDPActBasic != 0) != (tc.wantFeatures&netdevXDPActBasic != 0) { + t.Error("basic-action bit did not survive the round trip") + } + }) + } +}