diff --git a/docs/agents/ARCHITECTURE-CNI.md b/docs/agents/ARCHITECTURE-CNI.md index 29bcb4be..2e321e9d 100644 --- a/docs/agents/ARCHITECTURE-CNI.md +++ b/docs/agents/ARCHITECTURE-CNI.md @@ -96,6 +96,29 @@ delivery reads the per-interface `ifindex_egress_kind_table` row, which is exact. Once no egress shard resolves, a pass skips the remaining VRFs' shard routes rather than wait out each one's neighbor solicitation. +The ingress sidecar (`galactic-vrf`) writes rows into the same maps from its +pod's network namespace, each in a key range no host writer reaches: `vrf_table` +under `uformat.BlockIngressSidecar`, `ifindex_vrf_table` at or above +`ifindexvrfmap.SidecarIfindexBase`, and `egress_route_table` for tables at or +above `vrf.SidecarTableIDBase`. It leaves them in place when it exits, so a +restarted container keeps serving, and the host's GC and repair passes never +touch them. Two things remove a deleted pod's rows +(`internal/plumbing/ebpf/sidecarmap`, #773). A new sidecar removes, at startup, +every sidecar row its own VRFs do not account for. The installer's sidecar +return sweep (`internal/installer/sidecarreap.go`) treats a sidecar gateway +advertisement whose recorded host-side interface no longer exists, or is no +longer a veth, as the advertisement of a deleted pod, and deletes it. When every +sidecar advertisement on the node is in that state on two sweeps in a row, it +first removes every sidecar row from the maps. The second sweep covers a +replacement pod that has written its rows but not yet republished its +advertisements. A node whose sidecar publishes no advertisements gives the +installer no evidence, so it keeps that node's rows. Either path can still take +a live sidecar's rows for a deleted pod's: the startup prune when two sidecar +pods overlap on a node, and the installer when a replacement pod fails to +publish for two sweeps. So every sidecar reads its own `vrf_table` and +`ifindex_vrf_table` rows back on each sweep and reapplies its VRFs and routes +when any are missing, the same reapply a datapath reload triggers. + `galactic-router`'s own reconciler independently derives the *same* SID value from the same inputs (`srv6.ComputeSID`, `internal/plumbing/srv6/usid.go`, called from `internal/reconcile/reconcile.go`) for the BGP control-plane diff --git a/internal/gc/gc.go b/internal/gc/gc.go index fc0be240..c4840e40 100644 --- a/internal/gc/gc.go +++ b/internal/gc/gc.go @@ -586,7 +586,10 @@ func SweepEBPFVRFTable(ctx context.Context, k8s client.Client, namespace, nodeNa // on the next tick, leaving vrf_table empty for those entries while // ifindex_vrf_table, which this sweep never touches, still holds them, and // silently blackholing the sidecar's outbound connections. Preserve every - // entry in that block unconditionally. + // entry in that block here. Removing a deleted sidecar pod's rows is the + // sidecar's own job at startup and the installer's task once every sidecar + // advertisement on the node has looked gone for two sweeps, since only they + // can tell a live sidecar from a gone one. existing, err := reg.VRF.List() if err != nil { slog.Error("GC: failed to list eBPF vrf_table for sweep", "err", err) diff --git a/internal/ingresssidecar/backend.go b/internal/ingresssidecar/backend.go index bf94b4da..e3e4e384 100644 --- a/internal/ingresssidecar/backend.go +++ b/internal/ingresssidecar/backend.go @@ -45,6 +45,16 @@ type Backend interface { // it. Store compares successive values to know when every VRF and route // must be reapplied. DatapathGeneration() (string, error) + // PruneDatapath removes every row in the node's shared eBPF maps that an + // ingress sidecar wrote for a VRF whose routing table is not in + // keepTableIDs, and reports how many it removed. Those rows belong to an + // earlier sidecar pod on this node, which never removes its own on exit. + PruneDatapath(keepTableIDs map[uint32]struct{}) (removed int, err error) + // DatapathRowsPresent reports whether the shared eBPF maps still hold the + // rows EnsureVRF registered for every VRF in tableIDs. Another writer can + // remove them without reloading the datapath, which DatapathGeneration + // does not see. + DatapathRowsPresent(tableIDs []uint32) (bool, error) } // VRFInfo describes one kernel VRF device discovered by Backend.ListVRFs. @@ -131,6 +141,14 @@ func (kernelBackend) DatapathGeneration() (string, error) { return datapathGeneration() } +func (kernelBackend) DatapathRowsPresent(tableIDs []uint32) (bool, error) { + return datapathRowsPresent(tableIDs) +} + +func (kernelBackend) PruneDatapath(keepTableIDs map[uint32]struct{}) (int, error) { + return pruneDatapath(keepTableIDs) +} + // vrfNameRegex matches the interface name generated for a VPC: a leading // letter, nine zero-padded base62 characters, and a trailing letter. // diff --git a/internal/ingresssidecar/controller.go b/internal/ingresssidecar/controller.go index 424073ff..e4b9d010 100644 --- a/internal/ingresssidecar/controller.go +++ b/internal/ingresssidecar/controller.go @@ -81,7 +81,11 @@ var ( ) // RunStartup blocks, seeding store from reader, taking its inventory of host -// state, then sweeping it every interval until ctx is done. +// state, removing the eBPF map rows an earlier sidecar pod left behind, then +// sweeping it every interval until ctx is done. +// +// The prune runs only after a successful inventory. A failed one can leave a +// VRF in this pod untracked, and pruning then would remove its live rows. // // Only a failed EndpointSlice list holds the sweep back, retried with backoff: // running Inventory without the seed would reopen the race SeedFromAPI closes. @@ -104,6 +108,11 @@ func RunStartup(ctx context.Context, reader client.Reader, store *Store, interva } if err := store.Inventory(ctx, time.Now()); err != nil { slog.Error("ingresssidecar: startup inventory", "error", err) + } else if err := store.PruneForeignDatapath(); err != nil { + // Not retried: the rows it leaves only hold map slots until the next + // sidecar pod's startup prune, or until the host's installer judges + // every sidecar on this node gone. + slog.Error("ingresssidecar: startup prune", "error", err) } RunSweeper(ctx, store, interval) } diff --git a/internal/ingresssidecar/controller_test.go b/internal/ingresssidecar/controller_test.go index 4927425d..fe1e62bd 100644 --- a/internal/ingresssidecar/controller_test.go +++ b/internal/ingresssidecar/controller_test.go @@ -154,6 +154,38 @@ func TestRunStartupSweepsAfterFailedSeed(t *testing.T) { }) } +// TestRunStartupPrunesAfterInventory verifies the startup prune runs once, +// after Inventory, so the VRFs the pod already holds are in its keep set. +func TestRunStartupPrunesAfterInventory(t *testing.T) { + c := fake.NewClientBuilder().WithScheme(newTestScheme(t)).Build() + + backend := newFakeBackend() + backend.seedRoute("existing", 9, mustPrefix(t, "fd00::9"), net.ParseIP("fd00:99::9")) + store := NewStore(backend, time.Hour, nil) + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { + defer close(done) + RunStartup(ctx, c, store, time.Hour) + }() + t.Cleanup(func() { + cancel() + <-done + }) + + waitFor(t, "startup prune", func() bool { + backend.mu.Lock() + defer backend.mu.Unlock() + return len(backend.pruned) == 1 + }) + backend.mu.Lock() + defer backend.mu.Unlock() + if _, ok := backend.pruned[0][9]; !ok { + t.Errorf("keep = %v, want it to hold the inventoried VRF's table 9", backend.pruned[0]) + } +} + func waitFor(t *testing.T, what string, cond func() bool) { t.Helper() deadline := time.Now().Add(5 * time.Second) diff --git a/internal/ingresssidecar/ebpfdatapath.go b/internal/ingresssidecar/ebpfdatapath.go index b567c67f..b4a4cfe0 100644 --- a/internal/ingresssidecar/ebpfdatapath.go +++ b/internal/ingresssidecar/ebpfdatapath.go @@ -20,6 +20,7 @@ import ( "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" "go.datum.net/galactic/internal/plumbing/ebpf/ifindexvrfmap" "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/sidecarmap" "go.datum.net/galactic/internal/plumbing/ebpf/uformat" "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" "go.datum.net/galactic/internal/plumbing/vrf" @@ -441,6 +442,69 @@ func removeEgressDatapath(tableID uint32) error { return errors.Join(errs...) } +// pruneDatapath removes every row in the shared maps that an ingress sidecar +// registered for a VRF whose routing table is not in keepTableIDs. +// +// This sidecar leaves its rows in place when it exits, so that a restarted +// container in the same pod finds them still serving. A deleted pod takes its +// VRFs and veths with it but leaves the rows, and its replacement only +// overwrites the rows of the VPCs it serves again. This sidecar is the only +// writer of the sidecar's share of these maps on its node, so whatever its own +// VRFs do not account for is a predecessor's. +func pruneDatapath(keepTableIDs map[uint32]struct{}) (int, error) { + keep := make(map[uint16]struct{}, len(keepTableIDs)) + for tableID := range keepTableIDs { + argument, err := argumentForTableID(tableID) + if err != nil { + continue // not a table this sidecar registers rows for + } + keep[argument] = struct{}{} + } + + maps, closer, err := sidecarmap.OpenPinned(ebpfPinDir) + if err != nil { + return 0, err + } + defer func() { _ = closer.Close() }() + + result, err := sidecarmap.Prune(maps, keep) + return result.Total(), err +} + +// datapathRowsPresent reports whether vrf_table and ifindex_vrf_table still hold +// the rows ensureEgressDatapath registered for each VRF in tableIDs. +// +// Those two are written together, before any egress_route_table entry, and +// removed together by every path that removes a sidecar's rows: this +// sidecar's own startup prune in a later pod, and the host installer's reaper +// once it judges every sidecar on the node gone. Either can be wrong about a +// live sidecar, in a pod overlapping its predecessor or one slow to publish +// its advertisements, and reading the rows back is what lets this one notice. +func datapathRowsPresent(tableIDs []uint32) (bool, error) { + if len(tableIDs) == 0 { + return true, nil + } + maps, closer, err := sidecarmap.OpenPinned(ebpfPinDir) + if err != nil { + return false, err + } + defer func() { _ = closer.Close() }() + + for _, tableID := range tableIDs { + argument, err := argumentForTableID(tableID) + if err != nil { + continue // not a table this sidecar registers rows for + } + if _, ok, err := maps.VRF.Get(ingressSidecarBlock, argument); err != nil || !ok { + return false, err + } + if _, ok, err := maps.Ifindex.Get(ifindexvrfmap.SidecarIfindex(argument)); err != nil || !ok { + return false, err + } + } + return true, nil +} + // ensureRedirectRoute installs a plain host route for prefix into this pod's // main routing table, out the VRF interface for tableID as a bare nexthop // device. A VRF master device needs no gateway address: naming it as the link diff --git a/internal/ingresssidecar/fakebackend_test.go b/internal/ingresssidecar/fakebackend_test.go index e611c132..0bf3c553 100644 --- a/internal/ingresssidecar/fakebackend_test.go +++ b/internal/ingresssidecar/fakebackend_test.go @@ -37,6 +37,19 @@ type fakeBackend struct { failEnsureRoute error failRemoveVRF error failRemoveRoute error + + // pruned records the keep set of every PruneDatapath call, and + // failPrune, if set, is the error it returns. + pruned []map[uint32]struct{} + failPrune error + + // rowsMissing, if set, is the set of VRF tables whose rows + // DatapathRowsPresent reports absent, standing in for another writer + // removing them. EnsureVRF on that table restores them. failRowsRead, + // if set, is the error DatapathRowsPresent returns. Reads are not + // logged in calls, since Store makes them on every Sweep. + rowsMissing map[uint32]bool + failRowsRead error } type routeRecord struct { @@ -60,6 +73,7 @@ func (f *fakeBackend) EnsureVRF(vpc string) (uint32, error) { return 0, f.failEnsureVRF } if id, ok := f.vrfs[vpc]; ok { + delete(f.rowsMissing, id) return id, nil } id := f.nextTableID @@ -132,6 +146,28 @@ func (f *fakeBackend) DatapathGeneration() (string, error) { return f.generation, f.failGeneration } +func (f *fakeBackend) PruneDatapath(keepTableIDs map[uint32]struct{}) (int, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls = append(f.calls, "PruneDatapath") + f.pruned = append(f.pruned, keepTableIDs) + return 0, f.failPrune +} + +func (f *fakeBackend) DatapathRowsPresent(tableIDs []uint32) (bool, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.failRowsRead != nil { + return false, f.failRowsRead + } + for _, id := range tableIDs { + if f.rowsMissing[id] { + return false, nil + } + } + return true, nil +} + // reloadDatapath simulates the CNI control daemon recreating the shared eBPF // maps empty: every route entry is lost and the generation moves to // testGenReloaded. diff --git a/internal/ingresssidecar/store.go b/internal/ingresssidecar/store.go index ee451fb9..c3d15fcf 100644 --- a/internal/ingresssidecar/store.go +++ b/internal/ingresssidecar/store.go @@ -395,10 +395,45 @@ func (s *Store) Inventory(ctx context.Context, now time.Time) error { return nil } +// PruneForeignDatapath removes the rows an earlier sidecar pod on this node left +// in the shared eBPF maps, keeping those of every VRF this one tracks. +// +// Call it once, after Inventory: by then every VRF in this pod's namespace is +// tracked, so a row it does not account for belongs to a pod that is gone. It +// holds the store's lock throughout, so a VRF SetDesired creates concurrently +// is either already tracked or registers its rows after the prune. +func (s *Store) PruneForeignDatapath() error { + s.mu.Lock() + defer s.mu.Unlock() + + keep := make(map[uint32]struct{}, len(s.vrfs)) + for _, v := range s.vrfs { + if v.tableID != 0 { + keep[v.tableID] = struct{}{} + } + } + removed, err := s.backend.PruneDatapath(keep) + if removed > 0 { + slog.Info("ingresssidecar: removed rows an earlier sidecar pod left in the shared eBPF maps", + "removed", removed) + } + if err != nil { + s.countError("prune_datapath") + return fmt.Errorf("prune an earlier sidecar pod's eBPF map rows: %w", err) + } + return nil +} + // checkDatapathLocked reapplies every live VRF and route when the backend's -// DatapathGeneration has changed since they were written, or when an earlier +// DatapathGeneration has changed since they were written, when another writer +// has removed this sidecar's rows from the shared maps, or when an earlier // reapply pass left something unapplied. Callers must hold s.mu. // +// Rows go missing without a reload when a sidecar pod starting on the same +// node, or the host installer's reaper, takes this live sidecar's rows for a +// deleted pod's (see PruneForeignDatapath and internal/installer/sidecarreap.go). +// Reading them back each sweep bounds that mistake to one sweep interval. +// // Nothing else would notice. The CNI control daemon reloads the shared eBPF // datapath independently of this sidecar, recreating its maps empty on a // schema change and re-pinning usid_egress on every load, while this sidecar @@ -417,7 +452,11 @@ func (s *Store) checkDatapathLocked() { return } if gen == s.generation && !s.reapplyPending { - return + if s.rowsPresentLocked() { + return + } + slog.Warn("ingresssidecar: this sidecar's rows are missing from the shared eBPF maps, reapplying") + s.reapplyPending = true } if s.generation == "" && !s.anyInstalledLocked() { s.generation = gen // nothing written yet, so nothing to reapply @@ -433,6 +472,25 @@ func (s *Store) checkDatapathLocked() { s.reapplyPending = !s.reapplyLocked() } +// rowsPresentLocked reports whether the shared eBPF maps still hold the rows of +// every VRF reapplyLocked would reapply. A failed read counts as present, for +// the same reason a failed generation read is not acted on. Callers must hold +// s.mu. +func (s *Store) rowsPresentLocked() bool { + var tableIDs []uint32 + for _, v := range s.vrfs { + if v.installed && v.absentSince.IsZero() && v.tableID != 0 { + tableIDs = append(tableIDs, v.tableID) + } + } + present, err := s.backend.DatapathRowsPresent(tableIDs) + if err != nil { + slog.Debug("ingresssidecar: read back eBPF map rows", "err", err) + return true + } + return present +} + // reapplyLocked re-runs EnsureVRF for every installed VRF with a live route, // then EnsureRoute for every installed route still desired, reporting whether // all of them succeeded. State within its teardown grace period is left alone: diff --git a/internal/ingresssidecar/store_test.go b/internal/ingresssidecar/store_test.go index 81d532de..11992fcb 100644 --- a/internal/ingresssidecar/store_test.go +++ b/internal/ingresssidecar/store_test.go @@ -8,6 +8,7 @@ import ( "context" "errors" "net" + "reflect" "testing" "time" ) @@ -536,3 +537,100 @@ func TestStoreSweepDropsNeverInstalledRoute(t *testing.T) { t.Error("route still tracked after its grace period") } } + +// TestStorePruneForeignDatapathKeepsEveryTrackedVRF verifies the startup prune +// keeps the rows of both a VRF a live EndpointSlice claimed and one Inventory +// found in the pod, since either may still be serving traffic. +func TestStorePruneForeignDatapathKeepsEveryTrackedVRF(t *testing.T) { + ctx := context.Background() + backend := newFakeBackend() + backend.seedRoute("leftover", 9, mustPrefix(t, "fd00::9"), net.ParseIP("fd00:99::9")) + store := NewStore(backend, testGrace, nil) + + desired := &DesiredRoute{VPC: testVPC1, Prefix: mustPrefix(t, "fd00::1"), SID: net.ParseIP("fd00:99::1")} + if err := store.SetDesired(ctx, "ns/pod-a", desired); err != nil { + t.Fatalf("SetDesired: %v", err) + } + if err := store.Inventory(ctx, time.Now()); err != nil { + t.Fatalf("Inventory: %v", err) + } + if err := store.PruneForeignDatapath(); err != nil { + t.Fatalf("PruneForeignDatapath: %v", err) + } + + if len(backend.pruned) != 1 { + t.Fatalf("PruneDatapath calls = %d, want 1", len(backend.pruned)) + } + want := map[uint32]struct{}{backend.vrfs[testVPC1]: {}, 9: {}} + if got := backend.pruned[0]; !reflect.DeepEqual(got, want) { + t.Errorf("keep = %v, want %v", got, want) + } +} + +// TestStorePruneForeignDatapathReportsFailure verifies a failed prune is +// returned and counted rather than swallowed. +func TestStorePruneForeignDatapathReportsFailure(t *testing.T) { + backend := newFakeBackend() + backend.failPrune = errTest + store := NewStore(backend, testGrace, nil) + + if err := store.PruneForeignDatapath(); !errors.Is(err, errTest) { + t.Errorf("PruneForeignDatapath = %v, want %v", err, errTest) + } +} + +// TestStoreSweepReappliesWhenRowsGoMissing covers another writer removing this +// live sidecar's rows without reloading the datapath, as a sidecar pod starting +// beside it or the host's reaper can: the next Sweep restores them. +func TestStoreSweepReappliesWhenRowsGoMissing(t *testing.T) { + ctx := context.Background() + backend := newFakeBackend() + backend.generation = testGenLoaded + store := NewStore(backend, testGrace, nil) + + desired := &DesiredRoute{VPC: testVPC1, Prefix: mustPrefix(t, "fd00::1"), SID: net.ParseIP("fd00:99::1")} + if err := store.SetDesired(ctx, "ns/pod-a", desired); err != nil { + t.Fatalf("SetDesired: %v", err) + } + store.Sweep(ctx, time.Now()) // records the generation + + backend.mu.Lock() + backend.rowsMissing = map[uint32]bool{backend.vrfs[testVPC1]: true} + backend.mu.Unlock() + + before := backend.callCount() + store.Sweep(ctx, time.Now()) + got := backend.callsSince(before) + want := []string{"EnsureVRF:" + testVPC1, "EnsureRoute:1/fd00::1/128"} + if !reflect.DeepEqual(got, want) { + t.Fatalf("calls after rows went missing = %v, want %v", got, want) + } + + before = backend.callCount() + store.Sweep(ctx, time.Now()) + if got := backend.callsSince(before); len(got) != 0 { + t.Errorf("calls once the rows are back = %v, want none", got) + } +} + +// TestStoreSweepRowsReadFailureDoesNothing verifies an unreadable map, as on a +// node whose datapath is not loaded, is not taken for missing rows. +func TestStoreSweepRowsReadFailureDoesNothing(t *testing.T) { + ctx := context.Background() + backend := newFakeBackend() + backend.generation = testGenLoaded + store := NewStore(backend, testGrace, nil) + + desired := &DesiredRoute{VPC: testVPC1, Prefix: mustPrefix(t, "fd00::1"), SID: net.ParseIP("fd00:99::1")} + if err := store.SetDesired(ctx, "ns/pod-a", desired); err != nil { + t.Fatalf("SetDesired: %v", err) + } + store.Sweep(ctx, time.Now()) + + backend.failRowsRead = errTest + before := backend.callCount() + store.Sweep(ctx, time.Now()) + if got := backend.callsSince(before); len(got) != 0 { + t.Errorf("calls with unreadable rows = %v, want none", got) + } +} diff --git a/internal/installer/installer.go b/internal/installer/installer.go index 2881cf51..5e0b8e68 100644 --- a/internal/installer/installer.go +++ b/internal/installer/installer.go @@ -510,6 +510,10 @@ type ebpfDatapathState struct { // datapath repair rebuilds a VRF's shard routes from exactly as CNI ADD // installs them. egress attachreg.EgressConfig + + // sidecarReap carries the ingress sidecar reaper's state from one sidecar + // return sweep to the next. Nil leaves reaping off. + sidecarReap *sidecarReaper } // startEBPFDatapath loads and attaches the eBPF datapath and returns the state @@ -528,7 +532,7 @@ func startEBPFDatapath(ctx context.Context, m *metrics.Metrics) (ebpfDatapathSta } slog.Info("eBPF uSID datapath loaded, pinned, and attached", "interfaces", ifaces, "pinDir", attach.PinDir) - state := ebpfDatapathState{ifaces: ifaces, watcher: watcher} + state := ebpfDatapathState{ifaces: ifaces, watcher: watcher, sidecarReap: &sidecarReaper{}} watcher.OnChange(rewriteEBPFInterfaces) // The closer is the loaded objects in production. A test fake stands in a diff --git a/internal/installer/sidecarreap.go b/internal/installer/sidecarreap.go new file mode 100644 index 00000000..0c4ad26b --- /dev/null +++ b/internal/installer/sidecarreap.go @@ -0,0 +1,271 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package installer + +import ( + "context" + "errors" + "fmt" + "io/fs" + "log/slog" + "slices" + "strings" + "sync" + + "github.com/vishvananda/netlink" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + "go.datum.net/galactic/internal/crdnames" + "go.datum.net/galactic/internal/plumbing/ebpf/attach" + "go.datum.net/galactic/internal/plumbing/ebpf/sidecarmap" + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +// Removing what a deleted ingress sidecar pod leaves behind. +// +// The sidecar deliberately leaves its state in place when it exits, so a +// restarted container finds it still serving. A deleted pod takes its network +// namespace with it but leaves two things the host can see: its gateway +// advertisements, and its rows in the node's shared eBPF maps (see +// internal/plumbing/ebpf/sidecarmap). A replacement pod overwrites only what it +// serves again, and on a node no sidecar returns to, nothing overwrites +// anything. +// +// Each advertisement records the host-side end of the pod's primary interface. +// That interface is deleted with the pod, which makes it the host's evidence +// that the sidecar is gone: no other host-visible object carries the pod's +// lifetime, and the host cannot enter the pod's namespace to look (see the +// package comment in sidecarreturn.go). + +// sidecarLiveness is what the host can tell about the sidecar that published +// one advertisement. +type sidecarLiveness int + +const ( + // sidecarLivenessUnknown covers an advertisement with no usable entry + // point recorded, and an interface lookup that failed for any reason but + // absence. Treated as live: nothing is removed on a guess. + sidecarLivenessUnknown sidecarLiveness = iota + sidecarLivenessLive + sidecarLivenessGone +) + +// ingressAdvertisement is one of this node's sidecar gateway advertisements, +// with what deleting it safely needs and the liveness of its publisher. +type ingressAdvertisement struct { + name string + uid types.UID + resourceVersion string + liveness sidecarLiveness +} + +// sidecarReapPlan is what one pass removes. +type sidecarReapPlan struct { + // gone are the advertisements whose sidecar pod is gone. + gone []ingressAdvertisement + // sidecarGone is true when every advertisement on this node is gone, so + // no sidecar on it is live and every sidecar row in the shared maps is a + // leftover. + sidecarGone bool +} + +// planSidecarReap decides what to remove from this node's ingress +// advertisements. +// +// A gone advertisement is always removed, even beside a live one: a pod that +// has been replaced leaves behind the advertisements of every VPC its +// replacement does not serve. +// +// The map rows are judged for the node as a whole, because nothing in them +// links a row to an advertisement. The sidecar keys its rows on its pod-netns +// table ID and its advertisements on the BGPVRFInstance VRFID, which are +// different numbers for the same VPC. So the rows are leftovers only when no +// advertisement is live or unknown. With no advertisements at all there is no +// evidence either way, as on a node whose sidecar publishes none, and they are +// kept. +func planSidecarReap(advs []ingressAdvertisement) sidecarReapPlan { + var plan sidecarReapPlan + for _, a := range advs { + if a.liveness == sidecarLivenessGone { + plan.gone = append(plan.gone, a) + } + } + plan.sidecarGone = len(plan.gone) > 0 && len(plan.gone) == len(advs) + return plan +} + +// fingerprint identifies the exact set of gone advertisements, by object +// identity and version, so two passes can tell whether they saw the same one. +func (p sidecarReapPlan) fingerprint() string { + keys := make([]string, 0, len(p.gone)) + for _, a := range p.gone { + keys = append(keys, string(a.uid)+"/"+a.resourceVersion) + } + slices.Sort(keys) + return strings.Join(keys, ",") +} + +// sidecarReaper removes a deleted sidecar pod's advertisements and map rows. +// The zero value is ready to use. +type sidecarReaper struct { + mu sync.Mutex + // pendingRows is the fingerprint of the previous pass's plan when that + // plan found the whole sidecar gone, and "" otherwise. + pendingRows string +} + +// hostLinkByIndexFn and pruneSidecarRowsFn indirect the netlink lookup and the +// pinned-map writes reap makes, so tests need neither a real interface nor a +// loaded datapath. +var ( + hostLinkByIndexFn = netlink.LinkByIndex + pruneSidecarRowsFn = func() (int, error) { return pruneAllSidecarRows(attach.PinDir) } +) + +// reap runs one pass. Advertisements whose sidecar is gone are deleted at once. +// The map rows are removed only once two passes in a row find the whole +// sidecar gone, with the same advertisements at the same versions. +// +// The second pass is what makes removing rows safe while a replacement pod +// starts. That pod writes its rows first and then republishes each +// advertisement with its own interface, so a single pass can land between the +// two and see only the old pod's interface. By the next pass the republish has +// changed the advertisement, and the fingerprint with it. +// +// The rows are removed before the advertisements, because deleting the +// advertisements removes the evidence: if the prune fails, the next pass sees +// the same plan and retries. +func (r *sidecarReaper) reap(ctx context.Context, k8s client.Client, namespace, nodeName string) error { + advs, err := ingressAdvertisements(ctx, k8s, namespace, nodeName) + if err != nil { + return err + } + plan := planSidecarReap(advs) + + r.mu.Lock() + defer r.mu.Unlock() + + gone := plan.gone + if plan.sidecarGone { + fp := plan.fingerprint() + if r.pendingRows != fp { + // First sighting. Keep the advertisements as the evidence the + // next pass confirms against. + r.pendingRows = fp + return nil + } + removed, err := pruneSidecarRowsFn() + if err != nil { + return fmt.Errorf("remove a deleted ingress sidecar's eBPF map rows: %w", err) + } + if removed > 0 { + slog.Info("Removed a deleted ingress sidecar's rows from the shared eBPF maps", "removed", removed) + } + } + r.pendingRows = "" + + var errs []error + for _, a := range gone { + if err := deleteIngressAdvertisement(ctx, k8s, namespace, a); err != nil { + errs = append(errs, err) + } + } + return errors.Join(errs...) +} + +// ingressAdvertisements returns this node's sidecar gateway advertisements +// with the liveness of the pod that published each. Selected the same way +// sidecarGatewayEndpoints selects them, but keeping the ones it skips, since +// an advertisement with no usable entry point still counts as unknown. +func ingressAdvertisements( + ctx context.Context, k8s client.Client, namespace, nodeName string, +) ([]ingressAdvertisement, error) { + list := &bgpv1alpha1.BGPAdvertisementList{} + if err := k8s.List(ctx, list, client.InNamespace(namespace)); err != nil { + return nil, fmt.Errorf("list BGPAdvertisements in namespace %s: %w", namespace, err) + } + + segment := crdnames.IngressAdvertisementSegment() + var out []ingressAdvertisement + for i := range list.Items { + a := &list.Items[i] + if a.Spec.RouterRef.Name != nodeName || !isIngressAdvertisementName(a.Name, nodeName, segment) { + continue + } + out = append(out, ingressAdvertisement{ + name: a.Name, + uid: a.UID, + resourceVersion: a.ResourceVersion, + liveness: sidecarLivenessOf(a.Annotations), + }) + } + return out, nil +} + +// sidecarLivenessOf reads the host-side interface an advertisement records and +// reports whether it still exists. +// +// Only a veth counts as the pod's interface. The kernel hands out interface +// indexes in increasing order and reuses one only after wrapping, so an index +// that now names some other kind of link has been reused, and the pod is gone. +func sidecarLivenessOf(annotations map[string]string) sidecarLiveness { + ifindex, _, ok := ingressEntryPointAnnotations(annotations) + if !ok { + return sidecarLivenessUnknown + } + link, err := hostLinkByIndexFn(ifindex) + if err != nil { + var notFound netlink.LinkNotFoundError + if errors.As(err, ¬Found) { + return sidecarLivenessGone + } + slog.Debug("Could not look up an ingress sidecar's host-side interface", "ifindex", ifindex, "err", err) + return sidecarLivenessUnknown + } + if link.Type() != "veth" { + return sidecarLivenessGone + } + return sidecarLivenessLive +} + +// deleteIngressAdvertisement deletes a only if it is still the object this +// pass judged. A sidecar republishing it in the meantime has changed its +// version, and the delete then fails its precondition and leaves it alone. +func deleteIngressAdvertisement( + ctx context.Context, k8s client.Client, namespace string, a ingressAdvertisement, +) error { + adv := &bgpv1alpha1.BGPAdvertisement{} + adv.Name = a.name + adv.Namespace = namespace + err := k8s.Delete(ctx, adv, client.Preconditions{UID: &a.uid, ResourceVersion: &a.resourceVersion}) + switch { + case err == nil: + slog.Info("Deleted the gateway advertisement of a deleted ingress sidecar pod", "advertisement", a.name) + return nil + case apierrors.IsNotFound(err), apierrors.IsConflict(err): + return nil + default: + return fmt.Errorf("delete gateway advertisement %s: %w", a.name, err) + } +} + +// pruneAllSidecarRows removes every ingress sidecar row from the shared eBPF +// maps pinned under pinDir. A datapath that is not loaded has no rows to +// remove. +func pruneAllSidecarRows(pinDir string) (int, error) { + maps, closer, err := sidecarmap.OpenPinned(pinDir) + if err != nil { + if errors.Is(err, fs.ErrNotExist) { + return 0, nil + } + return 0, err + } + defer func() { _ = closer.Close() }() + + result, err := sidecarmap.Prune(maps, nil) + return result.Total(), err +} diff --git a/internal/installer/sidecarreap_test.go b/internal/installer/sidecarreap_test.go new file mode 100644 index 00000000..3bd7909a --- /dev/null +++ b/internal/installer/sidecarreap_test.go @@ -0,0 +1,293 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package installer + +import ( + "context" + "errors" + "strconv" + "testing" + + "github.com/vishvananda/netlink" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + + "go.datum.net/galactic/internal/crdnames" + bgpv1alpha1 "go.datum.net/network/api/v1alpha1" +) + +// Host-side interface indexes the tests' advertisements record: one that names +// a live veth and one whose interface is gone. +const ( + liveHostIfindex = 24 + goneHostIfindex = 31 +) + +// stubHostLinks makes hostLinkByIndexFn report liveHostIfindex as a veth and +// every other index as absent. +func stubHostLinks(t *testing.T) { + t.Helper() + orig := hostLinkByIndexFn + t.Cleanup(func() { hostLinkByIndexFn = orig }) + hostLinkByIndexFn = func(index int) (netlink.Link, error) { + if index == liveHostIfindex { + return &netlink.Veth{LinkAttrs: netlink.LinkAttrs{Index: index}}, nil + } + return nil, netlink.LinkNotFoundError{} + } +} + +// stubPrune counts prunes, failing each with failWith when it is non-nil. +func stubPrune(t *testing.T, failWith *error) *int { + t.Helper() + orig := pruneSidecarRowsFn + t.Cleanup(func() { pruneSidecarRowsFn = orig }) + calls := 0 + pruneSidecarRowsFn = func() (int, error) { + calls++ + if failWith != nil && *failWith != nil { + return 0, *failWith + } + return 3, nil + } + return &calls +} + +// sidecarAdv is a sidecar gateway advertisement on testSidecarNode whose +// recorded host-side interface is hostIfindex. +func sidecarAdv(vpc string, hostIfindex int) *bgpv1alpha1.BGPAdvertisement { + adv := gatewayAdv(vpc, testSidecarNode, testSidecarAddr+"/128", 1) + adv.UID = types.UID("uid-" + vpc) + adv.Annotations[crdnames.AnnotationIngressHostIfindex] = strconv.Itoa(hostIfindex) + return adv +} + +func advExists(t *testing.T, c client.Client, adv *bgpv1alpha1.BGPAdvertisement) bool { + t.Helper() + got := &bgpv1alpha1.BGPAdvertisement{} + err := c.Get(context.Background(), client.ObjectKeyFromObject(adv), got) + if apierrors.IsNotFound(err) { + return false + } + if err != nil { + t.Fatalf("Get %s: %v", adv.Name, err) + } + return true +} + +func TestPlanSidecarReap(t *testing.T) { + live := ingressAdvertisement{name: "live", liveness: sidecarLivenessLive} + unknown := ingressAdvertisement{name: "unknown", liveness: sidecarLivenessUnknown} + gone := ingressAdvertisement{name: "gone", liveness: sidecarLivenessGone} + gone2 := ingressAdvertisement{name: "gone2", liveness: sidecarLivenessGone} + + tests := []struct { + name string + advs []ingressAdvertisement + wantGone int + wantSidecarGone bool + }{ + {name: "no advertisements is no evidence", advs: nil}, + {name: "a live sidecar", advs: []ingressAdvertisement{live}}, + {name: "every advertisement gone", advs: []ingressAdvertisement{gone, gone2}, wantGone: 2, wantSidecarGone: true}, + {name: "a replaced pod's leftovers beside a live one", advs: []ingressAdvertisement{live, gone}, wantGone: 1}, + {name: "unknown keeps the rows", advs: []ingressAdvertisement{unknown, gone}, wantGone: 1}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + plan := planSidecarReap(tt.advs) + if len(plan.gone) != tt.wantGone { + t.Errorf("gone = %d advertisements, want %d", len(plan.gone), tt.wantGone) + } + if plan.sidecarGone != tt.wantSidecarGone { + t.Errorf("sidecarGone = %t, want %t", plan.sidecarGone, tt.wantSidecarGone) + } + }) + } +} + +func TestSidecarLivenessOf(t *testing.T) { + orig := hostLinkByIndexFn + t.Cleanup(func() { hostLinkByIndexFn = orig }) + + annotated := func(ifindex int) map[string]string { + return sidecarAdv("2", ifindex).Annotations + } + tests := []struct { + name string + annotations map[string]string + link netlink.Link + err error + want sidecarLiveness + }{ + {name: "no entry point recorded", annotations: nil, want: sidecarLivenessUnknown}, + {name: "veth present", annotations: annotated(7), link: &netlink.Veth{}, want: sidecarLivenessLive}, + {name: "interface absent", annotations: annotated(7), err: netlink.LinkNotFoundError{}, want: sidecarLivenessGone}, + { + name: "index reused by another kind of link", annotations: annotated(7), + link: &netlink.Dummy{}, want: sidecarLivenessGone, + }, + {name: "lookup failed", annotations: annotated(7), err: errors.New("netlink busy"), want: sidecarLivenessUnknown}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + hostLinkByIndexFn = func(int) (netlink.Link, error) { return tt.link, tt.err } + if got := sidecarLivenessOf(tt.annotations); got != tt.want { + t.Errorf("sidecarLivenessOf() = %d, want %d", got, tt.want) + } + }) + } +} + +// TestSidecarReaper_GoneSidecarIsRemovedOnTheSecondPass covers a sidecar pod +// deleted with no replacement: the first pass only records what it saw, and +// the second removes the map rows and then the advertisements. +func TestSidecarReaper_GoneSidecarIsRemovedOnTheSecondPass(t *testing.T) { + stubHostLinks(t) + prunes := stubPrune(t, nil) + advA, advB := sidecarAdv("2", goneHostIfindex), sidecarAdv("3", goneHostIfindex) + c := newAdvClient(t, advA, advB) + r := &sidecarReaper{} + + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("first reap: %v", err) + } + if *prunes != 0 || !advExists(t, c, advA) || !advExists(t, c, advB) { + t.Fatalf("first pass removed something (prunes=%d); want it only to record the evidence", *prunes) + } + + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("second reap: %v", err) + } + if *prunes != 1 { + t.Errorf("prunes = %d, want 1", *prunes) + } + if advExists(t, c, advA) || advExists(t, c, advB) { + t.Error("a gone sidecar's advertisements survived the second pass") + } +} + +// TestSidecarReaper_ReplacedPodsLeftoversGoAtOnce covers a replaced pod: its +// replacement is live, so the leftover advertisement goes on the first pass +// and no map row is touched, since the live sidecar removes its predecessor's. +func TestSidecarReaper_ReplacedPodsLeftoversGoAtOnce(t *testing.T) { + stubHostLinks(t) + prunes := stubPrune(t, nil) + live, leftover := sidecarAdv("2", liveHostIfindex), sidecarAdv("3", goneHostIfindex) + c := newAdvClient(t, live, leftover) + r := &sidecarReaper{} + + for range 2 { + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("reap: %v", err) + } + } + if *prunes != 0 { + t.Errorf("prunes = %d, want none beside a live sidecar", *prunes) + } + if !advExists(t, c, live) { + t.Error("the live sidecar's advertisement was deleted") + } + if advExists(t, c, leftover) { + t.Error("the replaced pod's leftover advertisement survived") + } +} + +// TestSidecarReaper_RepublishBetweenPassesKeepsTheRows covers a replacement +// pod that wrote its rows just before the first pass and republished just +// after it. The advertisement then differs on the second pass, which must not +// remove the rows the new pod already relies on. +func TestSidecarReaper_RepublishBetweenPassesKeepsTheRows(t *testing.T) { + stubHostLinks(t) + prunes := stubPrune(t, nil) + adv := sidecarAdv("2", goneHostIfindex) + c := newAdvClient(t, adv) + r := &sidecarReaper{} + + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("first reap: %v", err) + } + + current := &bgpv1alpha1.BGPAdvertisement{} + if err := c.Get(context.Background(), client.ObjectKeyFromObject(adv), current); err != nil { + t.Fatal(err) + } + current.Annotations[crdnames.AnnotationIngressHostIfindex] = strconv.Itoa(liveHostIfindex) + if err := c.Update(context.Background(), current); err != nil { + t.Fatal(err) + } + + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("second reap: %v", err) + } + if *prunes != 0 { + t.Errorf("prunes = %d, want none once the sidecar republished", *prunes) + } + if !advExists(t, c, adv) { + t.Error("the republished advertisement was deleted") + } +} + +// TestSidecarReaper_FailedPruneKeepsTheEvidence verifies the advertisements +// stay when the rows could not be removed, so the next pass retries. +func TestSidecarReaper_FailedPruneKeepsTheEvidence(t *testing.T) { + stubHostLinks(t) + failWith := errors.New("map busy") + prunes := stubPrune(t, &failWith) + adv := sidecarAdv("2", goneHostIfindex) + c := newAdvClient(t, adv) + r := &sidecarReaper{} + + _ = r.reap(context.Background(), c, testNamespace, testSidecarNode) + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); !errors.Is(err, failWith) { + t.Fatalf("second reap = %v, want %v", err, failWith) + } + if !advExists(t, c, adv) { + t.Fatal("advertisement deleted although its rows were not removed") + } + + failWith = nil + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("retry: %v", err) + } + if *prunes != 2 || advExists(t, c, adv) { + t.Errorf("after the retry prunes = %d and advertisement present = %t, want 2 and false", + *prunes, advExists(t, c, adv)) + } +} + +// TestSidecarReaper_LeavesOtherAdvertisementsAlone verifies a pod attachment's +// advertisement and another node's sidecar advertisement are never judged, +// even when the interface they record is gone on this host. +func TestSidecarReaper_LeavesOtherAdvertisementsAlone(t *testing.T) { + stubHostLinks(t) + prunes := stubPrune(t, nil) + pod := podAdv("2", "att1", testSidecarNode, testSidecarAddr+"/128", 1) + pod.Annotations[crdnames.AnnotationIngressHostIfindex] = strconv.Itoa(goneHostIfindex) + other := gatewayAdv("2", "worker-other", testSidecarAddr+"/128", 1) + other.Annotations[crdnames.AnnotationIngressHostIfindex] = strconv.Itoa(goneHostIfindex) + c := newAdvClient(t, pod, other) + r := &sidecarReaper{} + + for range 2 { + if err := r.reap(context.Background(), c, testNamespace, testSidecarNode); err != nil { + t.Fatalf("reap: %v", err) + } + } + if *prunes != 0 || !advExists(t, c, pod) || !advExists(t, c, other) { + t.Errorf("reaper acted on advertisements it does not own (prunes=%d)", *prunes) + } +} + +// TestPruneAllSidecarRows_NoDatapathIsQuiet verifies a node whose datapath is +// not loaded, so nothing is pinned, reports nothing to remove rather than an +// error every sweep. +func TestPruneAllSidecarRows_NoDatapathIsQuiet(t *testing.T) { + removed, err := pruneAllSidecarRows(t.TempDir()) + if err != nil || removed != 0 { + t.Errorf("pruneAllSidecarRows() = %d, %v; want 0, nil", removed, err) + } +} diff --git a/internal/installer/sidecarreturn.go b/internal/installer/sidecarreturn.go index 6160a063..43796f3a 100644 --- a/internal/installer/sidecarreturn.go +++ b/internal/installer/sidecarreturn.go @@ -118,6 +118,13 @@ func reconcileSidecarReturnPath(ctx context.Context, st ebpfDatapathState) { if st.k8sClient == nil || st.nodeName == "" { return } + // Before the return path, so a deleted pod's advertisements are gone by + // the time it lists them, rather than failing to install every tick. + if st.sidecarReap != nil { + if err := st.sidecarReap.reap(ctx, st.k8sClient, st.namespace, st.nodeName); err != nil { + slog.Warn("Could not remove what a deleted ingress sidecar pod left behind", "err", err) + } + } if err := ensureSidecarReturnPath(ctx, st.k8sClient, st.namespace, st.nodeName); err != nil { slog.Warn("Could not install the ingress sidecar's return path; "+ "replies to this node's sidecar gateway addresses will be dropped until this succeeds", "err", err) diff --git a/internal/plumbing/ebpf/sidecarmap/prune.go b/internal/plumbing/ebpf/sidecarmap/prune.go new file mode 100644 index 00000000..9263abe7 --- /dev/null +++ b/internal/plumbing/ebpf/sidecarmap/prune.go @@ -0,0 +1,173 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +// Package sidecarmap removes the rows the ingress sidecar (internal/ +// ingresssidecar) writes into the node's shared, pinned eBPF maps. +// +// The sidecar writes those rows from inside its pod's network namespace and +// deliberately leaves them in place when it exits, so a sidecar restarting in +// the same pod picks its datapath back up without a blackout. A pod that is +// deleted, though, takes its VRFs and veths with it and leaves the rows behind, +// and nothing else ever removes them: the host's sweeps leave the sidecar's +// share of every map alone, since they cannot read the pod's namespace. +// +// Every row the sidecar writes is recognisable from its key alone, because each +// map is split into ranges no host writer reaches: +// +// - vrf_table rows under uformat.BlockIngressSidecar +// - ifindex_vrf_table rows at or above ifindexvrfmap.SidecarIfindexBase +// - egress_route_table rows for routing tables at or above +// vrf.SidecarTableIDBase +// +// Each range also encodes the VRF's uSID Argument, which is what a caller names +// the rows to keep by. +package sidecarmap + +import ( + "errors" + "fmt" + "io" + "net" + + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/ifindexvrfmap" + "go.datum.net/galactic/internal/plumbing/ebpf/uformat" + "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" + "go.datum.net/galactic/internal/plumbing/vrf" +) + +// Maps are the three maps the ingress sidecar writes rows into. +type Maps struct { + VRF *usidmap.VRFTable + Ifindex *ifindexvrfmap.IfindexVRFTable + EgressRoute *egressroutemap.EgressRouteTable +} + +// OpenPinned opens the three maps from their pins under pinDir. The returned +// closer releases this process's handles and must be closed when the caller is +// done; the maps stay pinned. +func OpenPinned(pinDir string) (Maps, io.Closer, error) { + registry, registryCloser, err := usidmap.OpenPinnedRegistry(pinDir) + if err != nil { + return Maps{}, nil, fmt.Errorf("sidecarmap: open pinned vrf_table: %w", err) + } + ifindexTable, ifindexCloser, err := ifindexvrfmap.OpenPinned(pinDir) + if err != nil { + _ = registryCloser.Close() + return Maps{}, nil, fmt.Errorf("sidecarmap: %w", err) + } + routeTable, routeCloser, err := egressroutemap.OpenPinnedEgressRouteTable(pinDir) + if err != nil { + _ = registryCloser.Close() + _ = ifindexCloser.Close() + return Maps{}, nil, fmt.Errorf("sidecarmap: %w", err) + } + m := Maps{VRF: registry.VRF, Ifindex: ifindexTable, EgressRoute: routeTable} + return m, closers{registryCloser, ifindexCloser, routeCloser}, nil +} + +// closers closes every handle OpenPinned opened. +type closers []io.Closer + +func (c closers) Close() error { + errs := make([]error, 0, len(c)) + for _, closer := range c { + errs = append(errs, closer.Close()) + } + return errors.Join(errs...) +} + +// Result counts the rows Prune removed from each map. +type Result struct { + VRFRows int + IfindexRows int + EgressRouteRows int +} + +// Total is the number of rows removed across every map. +func (r Result) Total() int { + return r.VRFRows + r.IfindexRows + r.EgressRouteRows +} + +// Prune removes every ingress sidecar row whose uSID Argument is not in keep. +// A nil or empty keep removes all of them. +// +// Host rows are never touched: each map is filtered to the sidecar's own key +// range before anything is deleted. Every map is attempted even if an earlier +// one fails, and the failures are joined. +func Prune(m Maps, keep map[uint16]struct{}) (Result, error) { + var ( + result Result + errs []error + ) + kept := func(argument uint16) bool { + _, ok := keep[argument] + return ok + } + + if entries, err := m.VRF.List(); err != nil { + errs = append(errs, err) + } else { + for _, e := range entries { + if e.Block != uformat.BlockIngressSidecar || kept(e.Argument) { + continue + } + if err := m.VRF.Unregister(e.Block, e.Argument); err != nil { + errs = append(errs, err) + continue + } + result.VRFRows++ + } + } + + if entries, err := m.Ifindex.List(); err != nil { + errs = append(errs, err) + } else { + for _, e := range entries { + if e.Ifindex < ifindexvrfmap.SidecarIfindexBase { + continue + } + // The index, not the row's value, says which Argument it is + // for: the sidecar creates each interface at exactly + // SidecarIfindex(argument). + offset := e.Ifindex - ifindexvrfmap.SidecarIfindexBase + if offset <= uint32(uformat.ArgumentMax) && kept(uint16(offset)) { + continue + } + if err := m.Ifindex.Unregister(e.Ifindex); err != nil { + errs = append(errs, err) + continue + } + result.IfindexRows++ + } + } + + if byTable, err := m.EgressRoute.Prefixes(); err != nil { + errs = append(errs, err) + } else { + for tableID, prefixes := range byTable { + if tableID < vrf.SidecarTableIDBase { + continue + } + offset := tableID - vrf.SidecarTableIDBase + if offset <= uint32(uformat.ArgumentMax) && kept(uint16(offset)) { + continue + } + for p := range prefixes { + _, prefix, err := net.ParseCIDR(p) + if err != nil { + errs = append(errs, fmt.Errorf("sidecarmap: parse egress_route_table prefix %q: %w", p, err)) + continue + } + if err := m.EgressRoute.Unregister(tableID, prefix); err != nil { + errs = append(errs, err) + continue + } + result.EgressRouteRows++ + } + } + } + + return result, errors.Join(errs...) +} diff --git a/internal/plumbing/ebpf/sidecarmap/prune_test.go b/internal/plumbing/ebpf/sidecarmap/prune_test.go new file mode 100644 index 00000000..9f0c8854 --- /dev/null +++ b/internal/plumbing/ebpf/sidecarmap/prune_test.go @@ -0,0 +1,217 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package sidecarmap + +import ( + "net" + "reflect" + "testing" + + "github.com/cilium/ebpf" + + "go.datum.net/galactic/internal/plumbing/ebpf/egressroutemap" + "go.datum.net/galactic/internal/plumbing/ebpf/ifindexvrfmap" + "go.datum.net/galactic/internal/plumbing/ebpf/prog" + "go.datum.net/galactic/internal/plumbing/ebpf/uformat" + "go.datum.net/galactic/internal/plumbing/ebpf/usidmap" + "go.datum.net/galactic/internal/plumbing/vrf" +) + +// memTable is an in-memory usidmap.Table keyed by any comparable key. +type memTable map[any]any + +func (m memTable) Put(key, value any) error { m[key] = value; return nil } + +func (m memTable) Lookup(key, valueOut any) error { + v, ok := m[key] + if !ok { + return ebpf.ErrKeyNotExist + } + reflect.ValueOf(valueOut).Elem().Set(reflect.ValueOf(v)) + return nil +} + +func (m memTable) Delete(key any) error { + if _, ok := m[key]; !ok { + return ebpf.ErrKeyNotExist + } + delete(m, key) + return nil +} + +func (m memTable) Iterate() usidmap.Iterator { + it := &memIterator{} + for k, v := range m { + it.keys = append(it.keys, k) + it.values = append(it.values, v) + } + return it +} + +// memIterator walks a snapshot of a memTable's entries. +type memIterator struct { + keys, values []any + idx int +} + +func (it *memIterator) Next(keyOut, valueOut any) bool { + if it.idx >= len(it.keys) { + return false + } + reflect.ValueOf(keyOut).Elem().Set(reflect.ValueOf(it.keys[it.idx])) + reflect.ValueOf(valueOut).Elem().Set(reflect.ValueOf(it.values[it.idx])) + it.idx++ + return true +} + +func (it *memIterator) Err() error { return nil } + +const ( + hostBlock = uint64(0xfd00_0000_0001) + hostArgument = uint16(7) + hostIfindex = uint32(12) + hostTableID = uint32(42) +) + +// fixture is one node's maps holding a host attachment's rows and the rows of +// two sidecar VRFs, with Arguments 1 and 2. +type fixture struct { + maps Maps + vrfRaw, ifxRaw, routeRaw memTable +} + +func newFixture(t *testing.T) fixture { + t.Helper() + f := fixture{vrfRaw: memTable{}, ifxRaw: memTable{}, routeRaw: memTable{}} + f.maps = Maps{ + VRF: usidmap.NewVRFTable(f.vrfRaw), + Ifindex: ifindexvrfmap.NewIfindexVRFTable(f.ifxRaw), + EgressRoute: egressroutemap.NewEgressRouteTable(f.routeRaw), + } + + if err := f.maps.VRF.Register(hostBlock, hostArgument, hostTableID, usidmap.EgressKindVeth); err != nil { + t.Fatal(err) + } + if err := f.maps.Ifindex.Register(hostIfindex, hostBlock, hostArgument); err != nil { + t.Fatal(err) + } + f.putRoute(hostTableID, "2001:db8:1::/64") + + for _, arg := range []uint16{1, 2} { + tableID := vrf.SidecarTableIDBase + uint32(arg) + if err := f.maps.VRF.Register(uformat.BlockIngressSidecar, arg, tableID, usidmap.EgressKindVeth); err != nil { + t.Fatal(err) + } + if err := f.maps.Ifindex.Register(ifindexvrfmap.SidecarIfindex(arg), uformat.BlockIngressSidecar, arg); err != nil { + t.Fatal(err) + } + f.putRoute(tableID, "2001:db8:2::5/128") + f.putRoute(tableID, "10.0.0.5/32") + } + return f +} + +// putRoute writes an egress_route_table row directly. Register resolves the +// SID's next hop over netlink, which a unit test has nothing to resolve +// against, and Prune only ever reads keys. +func (f fixture) putRoute(tableID uint32, cidr string) { + _, prefix, err := net.ParseCIDR(cidr) + if err != nil { + panic(err) + } + ones, _ := prefix.Mask.Size() + key := prog.UsidEgressRouteKey{TableId: tableID, Prefixlen: uint32(8*(4+1) + ones)} + if v4 := prefix.IP.To4(); v4 != nil { + key.Family = 1 + copy(key.Addr[:4], v4) + } else { + copy(key.Addr[:], prefix.IP.To16()) + } + f.routeRaw[key] = prog.UsidEgressRouteValue{} +} + +func (f fixture) routeTables(t *testing.T) map[uint32]int { + t.Helper() + byTable, err := f.maps.EgressRoute.Prefixes() + if err != nil { + t.Fatal(err) + } + out := map[uint32]int{} + for tableID, prefixes := range byTable { + out[tableID] = len(prefixes) + } + return out +} + +func TestPrune_EmptyKeepRemovesEverySidecarRowAndNoHostRow(t *testing.T) { + f := newFixture(t) + + result, err := Prune(f.maps, nil) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if want := (Result{VRFRows: 2, IfindexRows: 2, EgressRouteRows: 4}); result != want { + t.Errorf("result = %+v, want %+v", result, want) + } + + vrfRows, err := f.maps.VRF.List() + if err != nil { + t.Fatal(err) + } + if len(vrfRows) != 1 || vrfRows[0].Block != hostBlock { + t.Errorf("vrf_table = %+v, want only the host row", vrfRows) + } + ifxRows, err := f.maps.Ifindex.List() + if err != nil { + t.Fatal(err) + } + if len(ifxRows) != 1 || ifxRows[0].Ifindex != hostIfindex { + t.Errorf("ifindex_vrf_table = %+v, want only the host row", ifxRows) + } + if got, want := f.routeTables(t), map[uint32]int{hostTableID: 1}; !reflect.DeepEqual(got, want) { + t.Errorf("egress_route_table rows per table = %v, want %v", got, want) + } +} + +func TestPrune_KeepsTheNamedArguments(t *testing.T) { + f := newFixture(t) + + result, err := Prune(f.maps, map[uint16]struct{}{2: {}}) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if want := (Result{VRFRows: 1, IfindexRows: 1, EgressRouteRows: 2}); result != want { + t.Errorf("result = %+v, want %+v", result, want) + } + + if _, ok, _ := f.maps.VRF.Get(uformat.BlockIngressSidecar, 1); ok { + t.Error("vrf_table still holds argument 1") + } + if _, ok, _ := f.maps.VRF.Get(uformat.BlockIngressSidecar, 2); !ok { + t.Error("vrf_table lost argument 2, which keep names") + } + if _, ok, _ := f.maps.Ifindex.Get(ifindexvrfmap.SidecarIfindex(1)); ok { + t.Error("ifindex_vrf_table still holds argument 1's interface") + } + if _, ok, _ := f.maps.Ifindex.Get(ifindexvrfmap.SidecarIfindex(2)); !ok { + t.Error("ifindex_vrf_table lost argument 2's interface, which keep names") + } + want := map[uint32]int{hostTableID: 1, vrf.SidecarTableIDBase + 2: 2} + if got := f.routeTables(t); !reflect.DeepEqual(got, want) { + t.Errorf("egress_route_table rows per table = %v, want %v", got, want) + } +} + +func TestPrune_KeepingEverythingRemovesNothing(t *testing.T) { + f := newFixture(t) + + result, err := Prune(f.maps, map[uint16]struct{}{1: {}, 2: {}}) + if err != nil { + t.Fatalf("Prune: %v", err) + } + if result.Total() != 0 { + t.Errorf("removed %+v, want nothing", result) + } +}