From 2ccd4c6af710df3703e846ac445612164d007997 Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 6 Oct 2026 22:55:54 -0400 Subject: [PATCH] fix: Re-attach usid_egress on existing links CNI ADD attaches usid_egress to an attachment's host-side veth or tap once, and the tc filter keeps its own reference to that program. When galactic-cni starts a new build, attach.Load pins a fresh usid_egress, but nothing moved existing attachments onto it. Every VM and pod kept running the previous build's egress program, bound to the previous copies of its maps. So a fix to usid_egress never reached a running workload until it was recreated. And when a map is recreated because its layout changed, as vip_xlat_table was in #768, existing attachments kept reading the old copy: rows galactic-router, a later CNI ADD or the egress refresh sweep wrote into the new map never reached them. In staging us-central-1 one VM was still on an older program after the 2026-10-07 rollout. attach.Start now calls ReattachEgress after attaching usid_ingress. It scans only veths and taps, the only links CNI ADD attaches to, finds the filter named galactic_usid_egress on each one's ingress hook, and replaces it in place with the new program, keeping its handle and priority so a changed priority setting cannot leave a duplicate. A filter already on the new program is skipped, and a link deleted during the scan, a pod or VM being torn down, is skipped rather than reported. A failure is logged, not fatal: an attachment left on the previous program still forwards, and failing startup would drop the node's whole ingress datapath. The new root test attaches a previous build's usid_egress to a veth and its usid_ingress to an uplink in a test namespace, loads a second build, and checks that only the egress filter moves to the new program and that a second pass changes nothing. Related to #774 Co-Authored-By: Claude Opus 5.5 --- internal/plumbing/ebpf/attach/attach.go | 12 ++ .../plumbing/ebpf/attach/egress_reattach.go | 94 ++++++++++++ .../ebpf/attach/egress_reattach_test.go | 144 ++++++++++++++++++ 3 files changed, 250 insertions(+) create mode 100644 internal/plumbing/ebpf/attach/egress_reattach.go create mode 100644 internal/plumbing/ebpf/attach/egress_reattach_test.go diff --git a/internal/plumbing/ebpf/attach/attach.go b/internal/plumbing/ebpf/attach/attach.go index bc037912..7abd09b9 100644 --- a/internal/plumbing/ebpf/attach/attach.go +++ b/internal/plumbing/ebpf/attach/attach.go @@ -87,6 +87,18 @@ func Start(pinDir string) (objs *prog.UsidObjects, ifaces []string, err error) { return nil, nil, fmt.Errorf("attach: %w", err) } + // Not fatal: an attachment left on the previous usid_egress still forwards, + // only without what changed since. Failing startup would drop the ingress + // datapath for the whole node over it. + replaced, err := ReattachEgress(objs.UsidEgress) + if err != nil { + slog.Error("attach: could not move every attachment onto the new usid_egress; "+ + "those keep running the previous program and maps until their next CNI ADD", + "replaced", replaced, "err", err) + } else if replaced > 0 { + slog.Info("attach: moved existing attachments onto the new usid_egress", "replaced", replaced) + } + return objs, ifaces, nil } diff --git a/internal/plumbing/ebpf/attach/egress_reattach.go b/internal/plumbing/ebpf/attach/egress_reattach.go new file mode 100644 index 00000000..c776c7d8 --- /dev/null +++ b/internal/plumbing/ebpf/attach/egress_reattach.go @@ -0,0 +1,94 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package attach + +import ( + "errors" + "fmt" + + "github.com/cilium/ebpf" + "github.com/vishvananda/netlink" + "golang.org/x/sys/unix" +) + +// ReattachEgress moves every usid_egress filter on this node's veths and taps +// onto program, the usid_egress just loaded, and returns how many filters it +// replaced. +// +// A filter is recognized by egressFilterName on a link's ingress hook and is +// replaced in place, keeping its handle and priority. A filter already running +// program is left alone, and a link deleted during the scan is skipped. Every +// link is attempted, and the failures are joined and returned together. +func ReattachEgress(program *ebpf.Program) (int, error) { + if program == nil { + return 0, errors.New("attach: program is nil") + } + info, err := program.Info() + if err != nil { + return 0, fmt.Errorf("attach: read usid_egress program info: %w", err) + } + wantID, ok := info.ID() + if !ok { + return 0, errors.New("attach: kernel did not report usid_egress's program id") + } + + links, err := listLinksFn() + if err != nil { + return 0, fmt.Errorf("attach: list links: %w", err) + } + + replaced := 0 + var errs []error + for _, link := range links { + // CNI ADD attaches usid_egress only to a veth or a tap, so nothing + // else needs a filter dump. + switch link.Type() { + case "veth", "tuntap": + default: + continue + } + filters, err := netlink.FilterList(link, netlink.HANDLE_MIN_INGRESS) + if err != nil { + if linkGone(err) { + continue + } + errs = append(errs, fmt.Errorf("interface %q: list ingress filters: %w", link.Attrs().Name, err)) + continue + } + for _, f := range filters { + bpf, ok := f.(*netlink.BpfFilter) + if !ok || bpf.Name != egressFilterName || uint32(bpf.Id) == uint32(wantID) { + continue + } + replacement := &netlink.BpfFilter{ + FilterAttrs: bpf.FilterAttrs, + Fd: program.FD(), + Name: egressFilterName, + DirectAction: true, + } + if err := netlink.FilterReplace(replacement); err != nil { + if linkGone(err) { + continue + } + errs = append(errs, fmt.Errorf("interface %q: replace usid_egress filter: %w", link.Attrs().Name, err)) + continue + } + replaced++ + } + } + return replaced, errors.Join(errs...) +} + +// linkGone reports whether err means the link was deleted after it was +// listed, which a pod or VM being torn down during the scan causes. That +// attachment needs no new program, so it is skipped rather than reported. +func linkGone(err error) bool { + var notFound netlink.LinkNotFoundError + return errors.As(err, ¬Found) || errors.Is(err, unix.ENODEV) || errors.Is(err, unix.ENOENT) +} + +// listLinksFn indirects netlink.LinkList so a test can confine ReattachEgress +// to its own namespace's links. +var listLinksFn = netlink.LinkList diff --git a/internal/plumbing/ebpf/attach/egress_reattach_test.go b/internal/plumbing/ebpf/attach/egress_reattach_test.go new file mode 100644 index 00000000..bda68eaf --- /dev/null +++ b/internal/plumbing/ebpf/attach/egress_reattach_test.go @@ -0,0 +1,144 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package attach + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/cilium/ebpf" + "github.com/containernetworking/plugins/pkg/ns" + "github.com/vishvananda/netlink" +) + +// A new load moves an attachment off the previous usid_egress and onto its +// own. Without that, the attachment keeps running the old program against the +// old copies of every map, and rows written after the reload never reach it. +// Only usid_egress filters are touched, and a second pass changes nothing. +func TestReattachEgress_MovesAttachmentsToTheNewProgram(t *testing.T) { + requireRoot(t) + + oldPin := filepath.Join("/sys/fs/bpf", fmt.Sprintf("galactic-reattach-old-%d", os.Getpid())) + newPin := filepath.Join("/sys/fs/bpf", fmt.Sprintf("galactic-reattach-new-%d", os.Getpid())) + t.Cleanup(func() { _ = os.RemoveAll(oldPin); _ = os.RemoveAll(newPin) }) + + nsObj, err := ns.TempNetNS() + if err != nil { + t.Fatalf("create test netns: %v", err) + } + defer func() { _ = nsObj.Close() }() + + err = nsObj.Do(func(_ ns.NetNS) error { + // tenant0 stands in for an attachment's host-side veth, and uplink0 + // for a shared uplink, which carries usid_ingress instead. + links := []netlink.Link{ + &netlink.Veth{LinkAttrs: netlink.LinkAttrs{Name: "tenant0"}, PeerName: "tenant0p"}, + &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "uplink0"}}, + } + for _, l := range links { + if err := netlink.LinkAdd(l); err != nil { + return fmt.Errorf("add %s: %w", l.Attrs().Name, err) + } + if err := netlink.LinkSetUp(l); err != nil { + return fmt.Errorf("set %s up: %w", l.Attrs().Name, err) + } + } + + previous, err := Load(oldPin) + if err != nil { + return fmt.Errorf("load previous build: %w", err) + } + defer func() { _ = previous.Close() }() + if err := AttachEgress(previous.UsidEgress, "tenant0"); err != nil { + return fmt.Errorf("attach previous usid_egress: %w", err) + } + if err := Attach(previous.UsidIngress, []string{"uplink0"}); err != nil { + return fmt.Errorf("attach previous usid_ingress: %w", err) + } + + current, err := Load(newPin) + if err != nil { + return fmt.Errorf("load current build: %w", err) + } + defer func() { _ = current.Close() }() + + replaced, err := ReattachEgress(current.UsidEgress) + if err != nil { + return fmt.Errorf("ReattachEgress: %w", err) + } + if replaced != 1 { + t.Errorf("first pass replaced %d filters, want 1", replaced) + } + if got, want := filterProgID(t, "tenant0", egressFilterName), progID(t, current.UsidEgress); got != want { + t.Errorf("tenant0's usid_egress filter runs program %d, want the current build's %d", got, want) + } + if got, want := filterProgID(t, "uplink0", filterName), progID(t, previous.UsidIngress); got != want { + t.Errorf("uplink0's usid_ingress filter runs program %d, want it untouched at %d", got, want) + } + + replaced, err = ReattachEgress(current.UsidEgress) + if err != nil { + return fmt.Errorf("second ReattachEgress: %w", err) + } + if replaced != 0 { + t.Errorf("second pass replaced %d filters, want 0", replaced) + } + if n := countFilters(t, "tenant0"); n != 1 { + t.Errorf("tenant0 has %d ingress filters after two passes, want 1", n) + } + return nil + }) + if err != nil { + t.Fatal(err) + } +} + +func progID(t *testing.T, p *ebpf.Program) uint32 { + t.Helper() + info, err := p.Info() + if err != nil { + t.Fatalf("program info: %v", err) + } + id, ok := info.ID() + if !ok { + t.Fatal("kernel did not report a program id") + } + return uint32(id) +} + +// filterProgID returns the program id of the named BPF filter on link's +// ingress hook, or 0 if there is none. +func filterProgID(t *testing.T, link, name string) uint32 { + t.Helper() + l, err := netlink.LinkByName(link) + if err != nil { + t.Fatalf("look up %s: %v", link, err) + } + filters, err := netlink.FilterList(l, netlink.HANDLE_MIN_INGRESS) + if err != nil { + t.Fatalf("list %s filters: %v", link, err) + } + for _, f := range filters { + if bpf, ok := f.(*netlink.BpfFilter); ok && bpf.Name == name { + return uint32(bpf.Id) + } + } + return 0 +} + +func countFilters(t *testing.T, link string) int { + t.Helper() + l, err := netlink.LinkByName(link) + if err != nil { + t.Fatalf("look up %s: %v", link, err) + } + filters, err := netlink.FilterList(l, netlink.HANDLE_MIN_INGRESS) + if err != nil { + t.Fatalf("list %s filters: %v", link, err) + } + return len(filters) +}