diff --git a/internal/ingresssidecar/backend.go b/internal/ingresssidecar/backend.go index 1bc5f157..2f3fa152 100644 --- a/internal/ingresssidecar/backend.go +++ b/internal/ingresssidecar/backend.go @@ -39,6 +39,11 @@ type Backend interface { // ListRoutes returns every seg6-encapsulated route currently installed // in tableID — the route half of the same startup-inventory step. ListRoutes(tableID uint32) ([]RouteInfo, error) + // DatapathGeneration returns an opaque value that changes whenever the + // shared eBPF datapath this sidecar writes into is reloaded out from under + // it. Store compares successive values to know when every VRF and route + // must be reapplied. + DatapathGeneration() (string, error) } // VRFInfo describes one kernel VRF device discovered by Backend.ListVRFs. @@ -117,6 +122,10 @@ func (kernelBackend) RemoveRoute(prefix *net.IPNet, tableID uint32) error { return nil } +func (kernelBackend) DatapathGeneration() (string, error) { + return datapathGeneration() +} + // 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/doc.go b/internal/ingresssidecar/doc.go index fe0bfe6d..dbd57d84 100644 --- a/internal/ingresssidecar/doc.go +++ b/internal/ingresssidecar/doc.go @@ -28,7 +28,9 @@ // - backend.go: the kernel-facing interface Store converges against, and its // production implementation. // - store.go: the mutex-protected, two-granularity, grace-period-aware -// reconciler at this package's core. +// reconciler at this package's core, which also reapplies every live VRF +// and route whenever the CNI control daemon reloads the shared eBPF +// datapath out from under it. // - metrics.go: Prometheus metrics. // - controller.go: the controller-runtime glue turning watch events into // desired-state updates, plus the startup inventory and periodic sweep diff --git a/internal/ingresssidecar/ebpfdatapath.go b/internal/ingresssidecar/ebpfdatapath.go index cad37193..a07cc1e6 100644 --- a/internal/ingresssidecar/ebpfdatapath.go +++ b/internal/ingresssidecar/ebpfdatapath.go @@ -19,6 +19,7 @@ import ( "go.datum.net/galactic/internal/plumbing/ebpf/attach" "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" @@ -329,6 +330,45 @@ func ensureEgressDatapath(vpc string, tableID uint32) error { return nil } +// datapathGeneration identifies the eBPF objects currently pinned under +// ebpfPinDir by the kernel IDs of usid_egress and egress_route_table. Both +// change when the CNI control daemon reloads the datapath: the program is +// re-pinned on every load, and the maps are recreated empty when their schema +// changes. Either change strands this sidecar's state, since its veths keep +// running the program they were attached with, reading the maps it was loaded +// against, and a map recreated empty has lost every entry written to it. +func datapathGeneration() (string, error) { + program, err := ebpf.LoadPinnedProgram(filepath.Join(ebpfPinDir, attach.UsidEgressPinName), nil) + if err != nil { + return "", fmt.Errorf("load pinned usid_egress program: %w", err) + } + defer func() { _ = program.Close() }() + progInfo, err := program.Info() + if err != nil { + return "", fmt.Errorf("read pinned usid_egress program info: %w", err) + } + progID, ok := progInfo.ID() + if !ok { + return "", errors.New("kernel does not report a program ID for usid_egress") + } + + routeTable, err := ebpf.LoadPinnedMap(filepath.Join(ebpfPinDir, prog.UsidMapEgressRouteTable), nil) + if err != nil { + return "", fmt.Errorf("load pinned %s: %w", prog.UsidMapEgressRouteTable, err) + } + defer func() { _ = routeTable.Close() }() + mapInfo, err := routeTable.Info() + if err != nil { + return "", fmt.Errorf("read pinned %s info: %w", prog.UsidMapEgressRouteTable, err) + } + mapID, ok := mapInfo.ID() + if !ok { + return "", fmt.Errorf("kernel does not report a map ID for %s", prog.UsidMapEgressRouteTable) + } + + return fmt.Sprintf("%d/%d", progID, mapID), nil +} + // removeEgressDatapath undoes ensureEgressDatapath's registrations for the VPC // whose VRF table is tableID. It runs before RemoveVRF deletes the VRF // interface, while the veth names can still be derived. diff --git a/internal/ingresssidecar/ebpfdatapath_integration_test.go b/internal/ingresssidecar/ebpfdatapath_integration_test.go index 75ff159c..4e402206 100644 --- a/internal/ingresssidecar/ebpfdatapath_integration_test.go +++ b/internal/ingresssidecar/ebpfdatapath_integration_test.go @@ -221,3 +221,35 @@ func TestEnsureEgressDatapath_AttachesToVethPeerNotVRF(t *testing.T) { t.Fatal(err) } } + +// TestDatapathGeneration_ChangesOnReload verifies the generation Store keys +// its reapply on moves when the CNI control daemon reloads the datapath: +// every attach.Load re-pins usid_egress, and a map recreated after a schema +// change gets a new kernel ID too. +func TestDatapathGeneration_ChangesOnReload(t *testing.T) { + requireRoot(t) + setUpTestPinDir(t) + + first, err := datapathGeneration() + if err != nil { + t.Fatalf("datapathGeneration: %v", err) + } + if again, err := datapathGeneration(); err != nil || again != first { + t.Fatalf("datapathGeneration with no reload = %q, %v; want %q", again, err, first) + } + + reloaded, err := attach.Load(ebpfPinDir) + if err != nil { + t.Fatalf("attach.Load (reload): %v", err) + } + t.Cleanup(func() { _ = reloaded.Close() }) + + second, err := datapathGeneration() + if err != nil { + t.Fatalf("datapathGeneration after reload: %v", err) + } + if second == first { + t.Fatalf("datapathGeneration after reload = %q, want it to differ from %q", second, first) + } + t.Logf("generation %s -> %s", first, second) +} diff --git a/internal/ingresssidecar/fakebackend_test.go b/internal/ingresssidecar/fakebackend_test.go index 10918238..e611c132 100644 --- a/internal/ingresssidecar/fakebackend_test.go +++ b/internal/ingresssidecar/fakebackend_test.go @@ -24,6 +24,13 @@ type fakeBackend struct { routes map[string]routeRecord // "vpc/prefix" -> record calls []string // ordered call log, for assertions + // generation is what DatapathGeneration returns, and failGeneration, if + // set, the error it returns instead. Tests change generation to simulate + // the shared eBPF datapath being reloaded. Reads are not logged in calls, + // since Store makes them on every Sweep. + generation string + failGeneration error + // failEnsureVRF/failEnsureRoute/failRemoveVRF/failRemoveRoute, if set, // make the matching method return this error instead of succeeding. failEnsureVRF error @@ -119,6 +126,22 @@ func (f *fakeBackend) ListRoutes(tableID uint32) ([]RouteInfo, error) { return infos, nil } +func (f *fakeBackend) DatapathGeneration() (string, error) { + f.mu.Lock() + defer f.mu.Unlock() + return f.generation, f.failGeneration +} + +// reloadDatapath simulates the CNI control daemon recreating the shared eBPF +// maps empty: every route entry is lost and the generation moves to +// testGenReloaded. +func (f *fakeBackend) reloadDatapath() { + f.mu.Lock() + defer f.mu.Unlock() + f.generation = testGenReloaded + f.routes = make(map[string]routeRecord) +} + // routeCount/vrfCount let tests assert on the fake's installed state // directly, independent of Store's own bookkeeping. func (f *fakeBackend) routeCount() int { diff --git a/internal/ingresssidecar/metrics.go b/internal/ingresssidecar/metrics.go index b0dbbef4..58aec1e8 100644 --- a/internal/ingresssidecar/metrics.go +++ b/internal/ingresssidecar/metrics.go @@ -18,6 +18,7 @@ type Metrics struct { RoutePending prometheus.Gauge ReconcileErrs *prometheus.CounterVec ReconcileTime prometheus.Histogram + Reapplies prometheus.Counter } // NewMetrics builds a fresh, unregistered Metrics. Call MustRegister once @@ -47,7 +48,8 @@ func NewMetrics() *Metrics { ReconcileErrs: prometheus.NewCounterVec(prometheus.CounterOpts{ Namespace: metricsNamespace, Name: "reconcile_errors_total", - Help: "Reconcile errors, by kind (ensure_vrf, ensure_route, remove_vrf, remove_route).", + Help: "Reconcile errors, by kind " + + "(ensure_vrf, ensure_route, remove_vrf, remove_route, reapply_vrf, reapply_route).", }, []string{"kind"}), ReconcileTime: prometheus.NewHistogram(prometheus.HistogramOpts{ Namespace: metricsNamespace, @@ -55,11 +57,17 @@ func NewMetrics() *Metrics { Help: "Time taken by each Store.SetDesired call.", Buckets: prometheus.DefBuckets, }), + Reapplies: prometheus.NewCounter(prometheus.CounterOpts{ + Namespace: metricsNamespace, + Name: "reapply_total", + Help: "Times every tracked VRF and route was reapplied after the shared eBPF datapath was reloaded.", + }), } } // MustRegister registers every metric this type owns against reg, panicking on // a duplicate. Callers do this once per process, at startup. func (m *Metrics) MustRegister(reg prometheus.Registerer) { - reg.MustRegister(m.VRFActive, m.RouteActive, m.VRFPending, m.RoutePending, m.ReconcileErrs, m.ReconcileTime) + reg.MustRegister(m.VRFActive, m.RouteActive, m.VRFPending, m.RoutePending, + m.ReconcileErrs, m.ReconcileTime, m.Reapplies) } diff --git a/internal/ingresssidecar/store.go b/internal/ingresssidecar/store.go index 23d803bb..f166f44e 100644 --- a/internal/ingresssidecar/store.go +++ b/internal/ingresssidecar/store.go @@ -63,6 +63,14 @@ type Store struct { routes map[string]*routeState vrfs map[string]*vrfState + + // generation is the backend's DatapathGeneration as of the last time every + // tracked VRF and route was known to be written against it, or empty + // before the first successful read. reapplyPending is set when a reapply + // pass left something unapplied, so the next Sweep retries it even though + // the generation has not moved again. + generation string + reapplyPending bool } // NewStore returns a Store that converges against backend, delaying teardown of @@ -162,6 +170,15 @@ func (s *Store) SetDesired(ctx context.Context, key string, desired *DesiredRout return nil } + // Read before the first write rather than after it: a reload landing + // between the two then shows up on the next Sweep as a changed generation + // instead of being mistaken for the one those writes went into. + if s.generation == "" { + if gen, gerr := s.backend.DatapathGeneration(); gerr == nil { + s.generation = gen + } + } + v, ok := s.vrfs[desired.VPC] if !ok { v = &vrfState{} @@ -210,12 +227,17 @@ func (s *Store) SetDesired(ctx context.Context, key string, desired *DesiredRout // grace, so the two timers can never overlap and a VPC is never torn down while // one of its routes might still come back. // +// Sweep first reapplies every live VRF and route if the shared eBPF datapath +// has been reloaded since they were written; see checkDatapathLocked. +// // Call this periodically, never reactively: VRF teardown is an aggregate // condition over many routes, not one watched object's transition. func (s *Store) Sweep(ctx context.Context, now time.Time) { s.mu.Lock() defer s.mu.Unlock() + s.checkDatapathLocked() + pendingRoutes, pendingVRFs := 0, 0 liveVPCs := make(map[string]struct{}, len(s.vrfs)) @@ -335,6 +357,103 @@ func (s *Store) Inventory(ctx context.Context, now time.Time) error { return nil } +// checkDatapathLocked reapplies every live VRF and route when the backend's +// DatapathGeneration has changed since they were written, or when an earlier +// reapply pass left something unapplied. Callers must hold s.mu. +// +// 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 +// writes kernel state only on an EndpointSlice change, and ensures a VRF's +// datapath only when first creating it. Without this, a route lost to a reload +// stays lost until this process restarts, and usid_egress's miss on it falls +// through to the VRF's default route, looping traffic back into the VRF. +// +// A failed generation read is not acted on: it means the datapath is not +// loaded at all, which nothing here can repair, and the next successful read +// after it is loaded again will differ and trigger the reapply. +func (s *Store) checkDatapathLocked() { + gen, err := s.backend.DatapathGeneration() + if err != nil { + slog.Debug("ingresssidecar: read eBPF datapath generation", "err", err) + return + } + if gen == s.generation && !s.reapplyPending { + return + } + if s.generation == "" && !s.anyInstalledLocked() { + s.generation = gen // nothing written yet, so nothing to reapply + return + } + + slog.Info("ingresssidecar: eBPF datapath changed, reapplying every VRF and route", + "previous", s.generation, "current", gen, "retry", s.reapplyPending) + if s.metrics != nil { + s.metrics.Reapplies.Inc() + } + s.generation = gen + s.reapplyPending = !s.reapplyLocked() +} + +// 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: +// reinstalling it would only extend the life of something already on its way +// out. Callers must hold s.mu. +func (s *Store) reapplyLocked() bool { + ok := true + failedVPCs := make(map[string]struct{}) + for vpc, v := range s.vrfs { + if !v.installed || !v.absentSince.IsZero() { + continue + } + tableID, err := s.backend.EnsureVRF(vpc) + if err != nil { + s.countError("reapply_vrf") + slog.Error("ingresssidecar: reapply VRF", "vpc", vpc, "error", err) + failedVPCs[vpc] = struct{}{} + ok = false + continue + } + if tableID != v.tableID { + slog.Warn("ingresssidecar: VRF table ID changed on reapply", "vpc", vpc, + "previous", v.tableID, "current", tableID) + v.tableID = tableID + } + } + + for key, r := range s.routes { + if !r.installed || !r.absentSince.IsZero() { + continue + } + if _, failed := failedVPCs[r.vpc]; failed { + continue // already counted against this pass; retried with its VRF + } + v, found := s.vrfs[r.vpc] + if !found { + slog.Error("ingresssidecar: reapply found route with no tracked VRF", "key", key, "vpc", r.vpc) + continue + } + if err := s.backend.EnsureRoute(r.prefix, r.sid, v.tableID); err != nil { + s.countError("reapply_route") + slog.Error("ingresssidecar: reapply route", "key", key, "vpc", r.vpc, "error", err) + ok = false + } + } + return ok +} + +// anyInstalledLocked reports whether this Store has written any VRF to the +// backend. Callers must hold s.mu. +func (s *Store) anyInstalledLocked() bool { + for _, v := range s.vrfs { + if v.installed { + return true + } + } + return false +} + // prefixClaimedElsewhereLocked reports whether some other tracked route still // needs the kernel state installed for r's VPC and prefix, either because it // is desired or because its own grace period has not elapsed. Callers must diff --git a/internal/ingresssidecar/store_test.go b/internal/ingresssidecar/store_test.go index 33037539..36a3a0f9 100644 --- a/internal/ingresssidecar/store_test.go +++ b/internal/ingresssidecar/store_test.go @@ -6,6 +6,7 @@ package ingresssidecar import ( "context" + "errors" "net" "testing" "time" @@ -304,3 +305,169 @@ func TestStoreSharedPrefixSurvivesSiblingTeardown(t *testing.T) { t.Errorf("vrfCount = %d, want 1", got) } } + +// testGenLoaded and testGenReloaded are fake DatapathGeneration values for a +// datapath before and after the CNI control daemon reloads it. +const ( + testGenLoaded = "1/1" + testGenReloaded = "2/2" +) + +// callsSince returns the backend calls logged after the first n. +func (f *fakeBackend) callsSince(n int) []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.calls[n:]...) +} + +func (f *fakeBackend) callCount() int { + f.mu.Lock() + defer f.mu.Unlock() + return len(f.calls) +} + +// TestStoreSweepReappliesAfterDatapathReload is issue #609's regression +// test: a reload that empties the shared eBPF maps while the route's +// EndpointSlice stays unchanged must be repaired by the next Sweep, not left +// missing until this process restarts. +func TestStoreSweepReappliesAfterDatapathReload(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) + } + + backend.reloadDatapath() + before := backend.callCount() + store.Sweep(ctx, time.Now()) + + got := backend.callsSince(before) + want := []string{"EnsureVRF:" + testVPC1, "EnsureRoute:1/fd00::1/128"} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Fatalf("calls after reload = %v, want %v", got, want) + } + if n := backend.routeCount(); n != 1 { + t.Errorf("routeCount after reapply = %d, want 1", n) + } +} + +// TestStoreSweepNoReapplyWhenGenerationStable verifies an unchanged +// datapath costs no backend writes on any Sweep. +func TestStoreSweepNoReapplyWhenGenerationStable(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) + } + + before := backend.callCount() + for range 3 { + store.Sweep(ctx, time.Now()) + } + if got := backend.callsSince(before); len(got) != 0 { + t.Errorf("calls with a stable generation = %v, want none", got) + } +} + +// TestStoreSweepGenerationReadFailureDoesNothing verifies an unreadable +// generation, meaning no datapath is loaded at all, neither reapplies nor +// disturbs the stored generation, so the reapply fires once it loads again. +func TestStoreSweepGenerationReadFailureDoesNothing(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) + } + + backend.reloadDatapath() + backend.failGeneration = errors.New("pinned map not found") + before := backend.callCount() + store.Sweep(ctx, time.Now()) + if got := backend.callsSince(before); len(got) != 0 { + t.Fatalf("calls with an unreadable generation = %v, want none", got) + } + + backend.failGeneration = nil + store.Sweep(ctx, time.Now()) + if n := backend.routeCount(); n != 1 { + t.Errorf("routeCount once the generation is readable again = %d, want 1", n) + } +} + +// TestStoreReapplySkipsRoutesInGrace verifies a reload does not reinstall a +// route already waiting out its teardown grace period. +func TestStoreReapplySkipsRoutesInGrace(t *testing.T) { + ctx := context.Background() + backend := newFakeBackend() + backend.generation = testGenLoaded + store := NewStore(backend, testGrace, nil) + + live := &DesiredRoute{VPC: testVPC1, Prefix: mustPrefix(t, "fd00::1"), SID: net.ParseIP("fd00:99::1")} + leaving := &DesiredRoute{VPC: testVPC1, Prefix: mustPrefix(t, "fd00::2"), SID: net.ParseIP("fd00:99::2")} + for key, d := range map[string]*DesiredRoute{"ns/pod-a": live, "ns/pod-b": leaving} { + if err := store.SetDesired(ctx, key, d); err != nil { + t.Fatalf("SetDesired %s: %v", key, err) + } + } + if err := store.SetDesired(ctx, "ns/pod-b", nil); err != nil { + t.Fatalf("SetDesired nil: %v", err) + } + + backend.reloadDatapath() + before := backend.callCount() + store.Sweep(ctx, time.Now()) + + for _, c := range backend.callsSince(before) { + if c == "EnsureRoute:1/fd00::2/128" { + t.Errorf("reapply reinstalled a route within its grace period: %v", backend.callsSince(before)) + } + } + if n := backend.routeCount(); n != 1 { + t.Errorf("routeCount after reapply = %d, want 1 (the live route only)", n) + } +} + +// TestStoreReapplyRetriesOnFailure verifies a reapply pass that fails is +// retried on the next Sweep even though the generation has not moved again. +func TestStoreReapplyRetriesOnFailure(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) + } + + backend.reloadDatapath() + backend.failEnsureRoute = errors.New("map write failed") + store.Sweep(ctx, time.Now()) + if n := backend.routeCount(); n != 0 { + t.Fatalf("routeCount after failed reapply = %d, want 0", n) + } + + backend.failEnsureRoute = nil + store.Sweep(ctx, time.Now()) + if n := backend.routeCount(); n != 1 { + t.Fatalf("routeCount after retried reapply = %d, want 1", n) + } + + before := backend.callCount() + store.Sweep(ctx, time.Now()) + if got := backend.callsSince(before); len(got) != 0 { + t.Errorf("calls once the retry succeeded = %v, want none", got) + } +}