diff --git a/cmd/galactic-router/root.go b/cmd/galactic-router/root.go index 42590898..a10fb479 100644 --- a/cmd/galactic-router/root.go +++ b/cmd/galactic-router/root.go @@ -491,7 +491,7 @@ func setupServiceRouteController( ctrl.Log.Error(err, "service route eBPF maps are not available yet; initialization will retry on reconcile") } - if err := (&controller.ServiceRoutePolicyReconciler{ + serviceRoutes := &controller.ServiceRoutePolicyReconciler{ Client: mgr.GetClient(), Scheme: mgr.GetScheme(), NodeName: nodeName, @@ -501,9 +501,13 @@ func setupServiceRouteController( FrontendEnabled: frontendEnabled, SpreadPairing: spreadPairing, Applied: make(map[types.NamespacedName]map[serviceroute.IntentSlot]serviceroute.RouteIntent), - }).SetupWithManager(mgr); err != nil { + } + if err := serviceRoutes.SetupWithManager(mgr); err != nil { return err } + if err := mgr.Add(linkWatch("service-route-links", serviceRoutes.InterfaceChanged)); err != nil { + return fmt.Errorf("setup service route interface watch: %w", err) + } producerHealth := &controller.ServiceProducerHealthReconciler{ Client: mgr.GetClient(), NodeName: nodeName, diff --git a/docs/router/architecture.md b/docs/router/architecture.md index b0dbfc4a..c5c21960 100644 --- a/docs/router/architecture.md +++ b/docs/router/architecture.md @@ -122,6 +122,7 @@ there instead. - **A VRF's routes follow its kernel table.** `applyVRFs` re-reads the table of every VRF it has applied on each reconcile, from one VRF listing. When the kernel VRF was recreated on another table, the VRF is applied again, the route-target index moves to the new table, and `withdrawMovedVRFRoutes` removes the routes of the VRF's best paths from the old table, deleting only entries that still point at the SID each path names (#1270). A VRF that is absent, or a listing that fails, keeps the table it has. A VPC's table normally survives recreation through its node reservation; see [VRF table IDs](../cni/architecture.md#vrf-table-ids). - **`ServiceVIPBinding`s are generated, one per node per backend.** `NetworkServiceBindingReconciler` (`internal/controller/networkservice_binding_controller.go`) runs on every node and writes a binding for each VIP of each accepted `NetworkService`, at most one per family (`ruleTranslatedVIPs`) and, for the IPv4 VIP, only while some gateway node's `/IPv4Served` condition reports it loaded (`ipv4LoadedOnSomeNode`; a status change to that signal passes `ruleBindingInputsChanged`, and the node deletes its IPv4 bindings once no gateway does), and each backend the rule's `backendSelector` places on this node, through the same `selectRuleBackends` the gateway uses, so the gateway never sends a backend's flows to a node with no binding planned for it. Workloads are IPv6 only, so both VIPs' backends are the selected attachments' IPv6 addresses: an IPv4 VIP's binding pairs the IPv4 `vipAddress` with an IPv6 `backendAddress`, which the backend node serves by translating the flows (see [ARCHITECTURE-CNI.md](../cni/architecture.md#key-design-decisions)). `ServiceVIPBindingReconciler` refuses an IPv4 `backendAddress` with `BindFailed`. Each node writes and deletes only its own bindings (labelled `galactic.datum.net/binding-node` and `app.kubernetes.io/managed-by: galactic-router`), so there is one writer per object and no leader election. A binding is owned by its rule, records the rule's `vpcRef` (which names the VRF its rows go in) and takes `egressKind` from the attachment's interface mode. `ServiceVIPBindingReconciler` programs a binding only when a `BGPAdvertisement` on this node's `BGPRouter` with that VRF's ID has a prefix containing `backendAddress`, and otherwise reports `BindFailed`; it watches advertisements on its own router so a binding written before its advertisement converges, and teardown skips the check. A backend serves only the oldest rule (by creation time, then namespace/name) selecting it on the same backend port and protocol, since its node translates the backend's replies back to only one VIP; a claim (`backendClaim`) carries the VIP's family, since a reply to an IPv4 client takes the IPv4 VIP's egress-translated row and any other reply the IPv6 VIP's egress row, so one backend socket answers for one VIP of each family, an IPv4 rule never competes with an IPv6 one, and two rules with IPv4 VIPs on one socket do; both sides compute that owner with `ruleBackendOwners` over every rule in the namespace, so a newer rule gets no binding for the backend and no flows for it from the gateway, and every rule in a namespace is re-queued when one of them changes. Rules compete only within a VPC, so the per-VPC `VPCAttachment` fan-out still reaches every rule an attachment change can affect. The node summarizes its bindings on the rule as `/BackendsBound` (`BackendsClaimed` names a backend an older rule serves); a rule whose spec cannot be built into bindings (a second IPv6 VIP, or a selector that does not parse) keeps the node's existing bindings, the same as a rule that loses `Accepted`, so an invalid edit does not drop the backend nodes' `vip_xlat_table` rows; a second IPv4 VIP does the same for the IPv4 family alone, writing no IPv4 binding and keeping the IPv4 ones it had while the IPv6 VIP's bindings stay current, and both report `InvalidRule`. Only the nodes that hold bindings for such a rule report `InvalidRule`, since the gateways already report the error and a condition from every node would be a status write per node per edit. Each rule reconcile lists only its own VPC's attachments, through the VPCAttachment→`status.vpc` field index (`listVPCAttachmentsInVPC`), and skips the cache's deep copy because nothing it does writes to them. Several backends of one VIP can share a node: the gateway writes each backend's slot into the uSID and `usid.c` keys the ingress row on it (#799). - **Each node probes its own backends and publishes their health.** For a `NetworkService` with `spec.healthCheck`, `BackendHealthReconciler` (`internal/controller/backendhealth_controller.go`) runs one TCP probe loop per backend on its node (`internal/healthprobe`), over the backend's host-side interface (the attachment's `status.hostInterface`) from that interface's link-local address, on `healthCheck.port` or the rule's `backendPort`. The link-local source is what lets the SYN-ACK back: `usid_egress` hands a reply to a link-local destination to the kernel before its VIP rewrite (see [ARCHITECTURE-CNI.md](../cni/architecture.md#key-design-decisions)). A decision moves after `unhealthyThreshold` consecutive failures or `healthyThreshold` consecutive successes; a probe that cannot be sent (no interface, no link-local address) counts toward `Unknown`/`ProbeUnavailable`, never toward `Unhealthy`. The decision goes into `status.health` of each of the backend's bindings, with the attachment UID and probed port, by a merge patch of that field alone and only when it changes, so it never races the `Bound` condition. Every binding of one backend shares one loop. A restarted router seeds each loop from the published decision when its attachment UID and port still match, so a restart changes no gateway's table. A rule losing its `healthCheck` clears `status.health` and stops the loops. The design is [docs/gateway/backend-health.md](../gateway/backend-health.md). +- **A consumer whose interface is not ready is held, not failed.** When a selected consumer's host interface is not on the node, or has no attachment identity yet, as when an instance is recreated under the same name, `EBPFRouteProgrammer.Apply` keeps the intent with nothing installed and `repairDesired` installs it once the interface is ready. galactic-router watches the node's interfaces (`linkWatch`) and requeues the policies whose applied intents use an interface when it changes, so a consumer that only lacked its interface is installed on the interface event. The CNI registers an interface's attachment identity after the interface appears, with no event of its own, so a policy that holds a consumer is also rechecked after one second, doubling to the 30 second resync while the hold lasts, and the reconciler logs when a policy starts holding. `replacePolicyLocked` tries every intent of a policy and returns the failures together, so one consumer that cannot be programmed does not hold back the rest. - **A service route event requeues only the policies it can change.** `ServiceRoutePolicyReconciler` has one worker and each reconcile reads every attachment, so requeueing every policy for every event in the cell keeps its queue full and a new consumer waits a whole pass before its route is installed. `internal/controller/serviceroute_requeue.go` maps a `VPCAttachment` event, old and new object, to the policies that select it as a consumer (selector and `consumerVPCRef`) or whose endpoint selects it as a producer; a `ServiceEndpoint` or `ServiceEndpointProducerReport` event to the policies routing to that endpoint; and a `VPC` event to the policies whose `consumerVPCRef` names it. A policy whose selector does not parse, or whose endpoint is not cached, is requeued for every attachment event. Policy changes, and `BGPRouter` or `BGPVRFInstance` spec changes, still requeue every policy, since conflicts and SIDs are decided across all of them. - **A `Spread` service endpoint's producers are probed by their own node, and a consumer's route reads a producer set.** `ServiceProducerHealthReconciler` (`internal/controller/serviceproducerhealth_controller.go`) runs one `internal/healthprobe` loop per producer on its node of each `ServiceEndpoint` with `spec.distribution: Spread`, to the producer attachment's first IPv6 address on `healthCheck.port`, and server-side applies that node's `ServiceEndpointProducerReport` for the endpoint (at most 16 entries, owned by the endpoint) under its own field manager (`galactic-router/`), so no object grows with the number of nodes. It deletes its report when it has no producers left, deletes reports of other nodes not renewed for 10 minutes, and removes its own entries from the endpoint's `status.producers`, where routers that predate reports write. Each node renews its entries every 15 seconds (`renewTime`); an entry not renewed for 60 seconds is expired, drains nothing, and ranks a remote producer after every currently reported one (`serviceplan.ProducerReportCurrent`). `ServiceRoutePolicyReconciler` requeues for the next expiry and ignores report and endpoint updates that only renew current entries. `ServiceRoutePolicyReconciler` reads them, through `serviceplan.WithProducerReports`, which keeps, for each producer and node, the more recently renewed of its report entry and its `status.producers` entry: for a Spread endpoint, `serviceplan.Compile` emits one `LocalSet` intent per consumer holding every producer on the consumer's node, `Unhealthy` ones marked draining unless all are, and, while one is currently `Healthy`, those with no current result (`Unknown` or unreported) marked draining too. With `PreferNodeLocal`, a consumer whose node's producers include one currently `Unhealthy` and none currently `Healthy` or with an expired report gets a remote producer instead, as one on a node with no producer does, when another node has a producer currently `Healthy`, `Unknown` or unreported (`selectProducers`, shared by `serviceplan.ProgrammingPlan`). `EBPFRouteProgrammer` writes it as a mode-3 `service_route_table` row plus a `service_producer_set_table` row under the same key; `usid_egress` pins each TCP flow in `service_flow_table` and picks a producer per UDP datagram (see [docs/router/private-service-routes.md](../private-services/README.md#several-producers-on-one-node)). A changed `LocalSet` intent is applied before the old one is removed, and the programmer keeps the newest of a key's producer sets in the map, so a health or membership change never drops the consumer's route. A datapath without the producer set map gets `serviceroute.ErrProducerSetsUnavailable` for `LocalSet` intents only. - **Surviving routers clean up after a departed node.** Only a node's own `NetworkServiceBindingReconciler` clears its `/BackendsBound` conditions and deletes its generated bindings, and only its own `ServiceVIPBindingReconciler` removes the teardown finalizer from bindings targeting it. So once a node's `Node` object is deleted, `NodeDepartureReconciler` (`internal/controller/node_departure_controller.go`) on every other router removes that node's `BackendsBound` condition from every `NetworkService`, deletes its generated bindings and removes the finalizer from every binding whose `targetRef.name` names it, hand-written ones included; the kernel rows went with the node. Without it the conditions stayed forever and a deleted rule's bindings stayed `Terminating`. Every Node create and delete maps to one sweep key, so the sweep runs on the initial Node list (startup), on every Node delete and every `--gc-interval`, and a router down during the delete still converges. A node missing from the cached Node list counts as departed only once an uncached `Get` returns NotFound. Every router sweeps, with no leader election: status writes and finalizer patches retry on conflict (the patch carries the resourceVersion) and NotFound counts as done. diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index 4680e06b..ff5e85bc 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -71,6 +71,17 @@ type ServiceRoutePolicyReconciler struct { // (noteConflicts). conflicted map[types.NamespacedName]bool requeue chan event.GenericEvent + + // devices maps each interface name an applied intent uses to the + // policies using it, so an interface event requeues them + // (InterfaceChanged). It has its own lock so the interface watch never + // waits on a reconcile. + devicesMu sync.Mutex + devices map[string]map[types.NamespacedName]struct{} + + // heldBackoff is the next recheck delay of each policy that holds a + // consumer for an interface that is not ready (retryHeld). + heldBackoff map[types.NamespacedName]time.Duration } // compile compiles policy for node with this router's options. @@ -84,6 +95,41 @@ func (r *ServiceRoutePolicyReconciler) compile(policy *networkv1alpha1.ServiceRo const serviceRouteMapResyncInterval = 30 * time.Second +// retryHeld rechecks a policy sooner while the programmer holds one of its +// consumers for an interface that is not ready, backing off from +// heldInterfaceRetry to serviceRouteMapResyncInterval so a consumer whose +// interface never returns costs one reconcile per resync. It logs when a +// policy starts holding. +func (r *ServiceRoutePolicyReconciler) retryHeld(logger logr.Logger, key types.NamespacedName, + intents []serviceroute.RouteIntent, result ctrl.Result, +) ctrl.Result { + holder, ok := r.Programmer.(interface { + Holding([]serviceroute.RouteIntent) bool + }) + if !ok || !holder.Holding(intents) { + delete(r.heldBackoff, key) + return result + } + if r.heldBackoff == nil { + r.heldBackoff = make(map[types.NamespacedName]time.Duration) + } + wait, held := r.heldBackoff[key] + if !held { + logger.Info("holding a service route until its interface is ready") + wait = heldInterfaceRetry + } + r.heldBackoff[key] = min(2*wait, serviceRouteMapResyncInterval) + if result.RequeueAfter == 0 || result.RequeueAfter > wait { + result.RequeueAfter = wait + } + return result +} + +// heldInterfaceRetry is the first recheck delay of a policy holding a consumer +// for an interface that has no attachment identity yet. An interface event +// installs one that only lacked the interface. +const heldInterfaceRetry = time.Second + // Startup sync retries back off from startupSyncInitialBackoff to // startupSyncMaxBackoff. A map that is not pinned yet logs at V(1) until it // has been missing for startupSyncMissingMapGrace, since the datapath loader @@ -296,7 +342,8 @@ func (r *ServiceRoutePolicyReconciler) reconcilePolicy( logger.Error(err, "update ServiceRoutePolicy acceptance status") return ctrl.Result{}, err } - return serviceRouteRequeue(policy, endpoint, intents, time.Now()), nil + result := serviceRouteRequeue(policy, endpoint, intents, time.Now()) + return r.retryHeld(logger, req.NamespacedName, intents, result), nil } // serviceRouteRequeue returns when a programmed policy is next checked: at the @@ -736,6 +783,7 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked( intents []serviceroute.RouteIntent, ) error { defer r.updateProgrammedGauge() + defer r.indexPolicyDevices(key) if r.Applied == nil { r.Applied = make(map[types.NamespacedName]map[serviceroute.IntentSlot]serviceroute.RouteIntent) } @@ -794,6 +842,9 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked( r.Metrics.ObserveOperation("remove", nil) delete(current, slot) } + // One consumer that cannot be programmed must not hold back the others, + // so every intent is tried and the failures are returned together. + var applyErrs []error for slot, intent := range desired { if existing, ok := current[slot]; ok && reflect.DeepEqual(existing, intent) { continue @@ -801,9 +852,10 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked( if err := r.Programmer.Apply(intent); err != nil { r.Metrics.ObserveOperation("apply", err) if cleanupErr := r.Programmer.Cleanup(intent); cleanupErr != nil { - return errors.Join(err, &serviceRoutePolicyCleanupError{err: cleanupErr}) + return errors.Join(append(applyErrs, err, &serviceRoutePolicyCleanupError{err: cleanupErr})...) } - return err + applyErrs = append(applyErrs, err) + continue } r.Metrics.ObserveOperation("apply", nil) current[slot] = intent @@ -811,7 +863,7 @@ func (r *ServiceRoutePolicyReconciler) replacePolicyLocked( if len(current) == 0 { delete(r.Applied, key) } - return nil + return errors.Join(applyErrs...) } func (r *ServiceRoutePolicyReconciler) removePolicy(key types.NamespacedName) error { @@ -822,6 +874,8 @@ func (r *ServiceRoutePolicyReconciler) removePolicy(key types.NamespacedName) er func (r *ServiceRoutePolicyReconciler) removePolicyLocked(key types.NamespacedName) error { defer r.updateProgrammedGauge() + defer r.indexPolicyDevices(key) + delete(r.heldBackoff, key) current := r.Applied[key] for attachment, intent := range current { if err := r.Programmer.Remove(intent); err != nil { diff --git a/internal/controller/serviceroute_controller_test.go b/internal/controller/serviceroute_controller_test.go index eeee92a4..ca98f1b0 100644 --- a/internal/controller/serviceroute_controller_test.go +++ b/internal/controller/serviceroute_controller_test.go @@ -10,13 +10,17 @@ import ( "fmt" "io/fs" "net" + "reflect" "testing" "time" + "github.com/go-logr/logr" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" @@ -506,3 +510,104 @@ func TestServiceRouteStartupSyncBacksOffWhileMapsMissing(t *testing.T) { }) } } + +// One consumer that cannot be programmed does not hold back the rest of the +// policy: every other intent is applied and the failure is still returned. +func TestServiceRouteReplacePolicyAppliesPastAFailingConsumer(t *testing.T) { + programmer := &retryRouteProgrammer{failAttachment: "b"} + reconciler := &ServiceRoutePolicyReconciler{Programmer: programmer} + policy := types.NamespacedName{Namespace: testServiceRoutePolicyNamespace, Name: testServiceRouteName} + intents := []serviceroute.RouteIntent{ + {Attachment: types.NamespacedName{Namespace: testServiceRouteTenantNamespace, Name: "a"}}, + {Attachment: types.NamespacedName{Namespace: testServiceRouteTenantNamespace, Name: "b"}}, + {Attachment: types.NamespacedName{Namespace: testServiceRouteTenantNamespace, Name: "c"}}, + } + for range 5 { + reconciler.Applied = nil + programmer.applyCalls = 0 + if err := reconciler.replacePolicy(policy, intents); err == nil { + t.Fatal("replacePolicy succeeded, want the failing consumer reported") + } + if programmer.applyCalls != 3 { + t.Fatalf("Apply calls = %d, want every intent tried", programmer.applyCalls) + } + if got := len(reconciler.Applied[policy]); got != 2 { + t.Fatalf("tracked intents = %d, want the two that succeeded", got) + } + } +} + +// An interface event requeues the policies whose applied intents use that +// interface, and only those. +func TestServiceRouteInterfaceChangedRequeuesPoliciesUsingIt(t *testing.T) { + reconciler := &ServiceRoutePolicyReconciler{ + Programmer: &retryRouteProgrammer{}, + requeue: make(chan event.GenericEvent, 8), + } + dns := types.NamespacedName{Namespace: testServiceRoutePolicyNamespace, Name: testServiceRouteName} + other := types.NamespacedName{Namespace: testServiceRoutePolicyNamespace, Name: "dns-tcp"} + if err := reconciler.replacePolicy(dns, []serviceroute.RouteIntent{{ + Attachment: types.NamespacedName{Namespace: testServiceRouteTenantNamespace, Name: "a"}, + ConsumerDevice: "tap-a", + }}); err != nil { + t.Fatal(err) + } + if err := reconciler.replacePolicy(other, []serviceroute.RouteIntent{{ + Attachment: types.NamespacedName{Namespace: testServiceRouteTenantNamespace, Name: "b"}, + ConsumerDevice: "tap-b", + }}); err != nil { + t.Fatal(err) + } + + reconciler.InterfaceChanged("tap-a") + if len(reconciler.requeue) != 1 || (<-reconciler.requeue).Object.GetName() != dns.Name { + t.Fatal("tap-a did not requeue exactly the dns policy") + } + reconciler.InterfaceChanged("unknown") + if len(reconciler.requeue) != 0 { + t.Fatal("an unused interface requeued a policy") + } + if err := reconciler.removePolicy(dns); err != nil { + t.Fatal(err) + } + reconciler.InterfaceChanged("tap-a") + if len(reconciler.requeue) != 0 { + t.Fatal("a removed policy was still requeued") + } +} + +type holdingProgrammer struct { + retryRouteProgrammer + holding bool +} + +func (p *holdingProgrammer) Holding([]serviceroute.RouteIntent) bool { return p.holding } + +// A policy holding a consumer is rechecked soon, backing off to the resync +// interval, and a policy that holds nothing keeps its own requeue. +func TestServiceRouteRetryHeldBacksOffPerPolicy(t *testing.T) { + programmer := &holdingProgrammer{holding: true} + reconciler := &ServiceRoutePolicyReconciler{Programmer: programmer} + key := types.NamespacedName{Namespace: testServiceRoutePolicyNamespace, Name: testServiceRouteName} + got := make([]time.Duration, 0, 7) + for range 7 { + got = append(got, reconciler.retryHeld(logr.Discard(), key, nil, ctrl.Result{}).RequeueAfter) + } + want := []time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, + 16 * time.Second, 30 * time.Second, 30 * time.Second} + if !reflect.DeepEqual(got, want) { + t.Fatalf("held requeues = %v, want %v", got, want) + } + programmer.holding = false + if r := reconciler.retryHeld(logr.Discard(), key, nil, ctrl.Result{}); r.RequeueAfter != 0 { + t.Fatalf("policy no longer holding requeued after %v, want its own result", r.RequeueAfter) + } + own := ctrl.Result{RequeueAfter: 30 * time.Second} + if r := reconciler.retryHeld(logr.Discard(), key, nil, own); r.RequeueAfter != own.RequeueAfter { + t.Fatalf("requeue = %v, want the policy's own 30s", r.RequeueAfter) + } + programmer.holding = true + if r := reconciler.retryHeld(logr.Discard(), key, nil, ctrl.Result{}); r.RequeueAfter != time.Second { + t.Fatalf("a new hold started at %v, want 1s", r.RequeueAfter) + } +} diff --git a/internal/controller/serviceroute_requeue.go b/internal/controller/serviceroute_requeue.go index 6b019792..b814dd96 100644 --- a/internal/controller/serviceroute_requeue.go +++ b/internal/controller/serviceroute_requeue.go @@ -20,6 +20,7 @@ import ( cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" + "go.datum.net/galactic/internal/serviceroute" ) // requestQueue is the controller's work queue. @@ -233,3 +234,67 @@ func (r *ServiceRoutePolicyReconciler) noteConflicts(self types.NamespacedName, } } } + +// intentDevices returns the interface names intent is programmed through. +func intentDevices(intent serviceroute.RouteIntent) []string { + devices := make([]string, 0, 2+len(intent.Producers)) + for _, name := range []string{intent.ConsumerDevice, intent.ServiceDevice} { + if name != "" { + devices = append(devices, name) + } + } + for _, producer := range intent.Producers { + if producer.Device != "" { + devices = append(devices, producer.Device) + } + } + return devices +} + +// indexPolicyDevices records which interfaces key's applied intents use. The +// caller holds r.mu. +func (r *ServiceRoutePolicyReconciler) indexPolicyDevices(key types.NamespacedName) { + r.devicesMu.Lock() + defer r.devicesMu.Unlock() + if r.devices == nil { + r.devices = make(map[string]map[types.NamespacedName]struct{}) + } + for name, policies := range r.devices { + delete(policies, key) + if len(policies) == 0 { + delete(r.devices, name) + } + } + for _, intent := range r.Applied[key] { + for _, name := range intentDevices(intent) { + if r.devices[name] == nil { + r.devices[name] = make(map[types.NamespacedName]struct{}) + } + r.devices[name][key] = struct{}{} + } + } +} + +// InterfaceChanged requeues every policy with an applied intent through the +// interface name, so a consumer held while its interface was missing, or a +// recreated interface, is programmed as soon as the interface appears. +func (r *ServiceRoutePolicyReconciler) InterfaceChanged(name string) { + r.devicesMu.Lock() + keys := make([]types.NamespacedName, 0, len(r.devices[name])) + for key := range r.devices[name] { + keys = append(keys, key) + } + r.devicesMu.Unlock() + if r.requeue == nil { + return + } + for _, key := range keys { + select { + case r.requeue <- event.GenericEvent{Object: &networkv1alpha1.ServiceRoutePolicy{ + ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name}, + }}: + default: + // Full: the policy's periodic requeue still reaches it. + } + } +} diff --git a/internal/plumbing/ebpf/serviceroutemap/table.go b/internal/plumbing/ebpf/serviceroutemap/table.go index 8451dde4..9e83191e 100644 --- a/internal/plumbing/ebpf/serviceroutemap/table.go +++ b/internal/plumbing/ebpf/serviceroutemap/table.go @@ -563,6 +563,10 @@ func (t *Tables) PolicyEnabled() (bool, error) { return value.Enabled != 0, nil } +// ErrNoAttachmentIdentity reports an interface the CNI has not registered an +// attachment identity for, either yet or any more. +var ErrNoAttachmentIdentity = errors.New("no attachment identity") + // IdentityToken returns the current nonzero incarnation token for ifindex. func (t *Tables) IdentityToken(ifindex uint32) (uint64, error) { if t.identities == nil { @@ -570,6 +574,10 @@ func (t *Tables) IdentityToken(ifindex uint32) (uint64, error) { } var token uint64 if err := t.identities.Lookup(ifindex, &token); err != nil { + if errors.Is(err, ebpf.ErrKeyNotExist) { + return 0, fmt.Errorf("serviceroutemap: lookup attachment identity ifindex=%d: %w: %w", + ifindex, ErrNoAttachmentIdentity, err) + } return 0, fmt.Errorf("serviceroutemap: lookup attachment identity ifindex=%d: %w", ifindex, err) } if token == 0 { diff --git a/internal/serviceroute/dataplane.go b/internal/serviceroute/dataplane.go index 8eec573c..ca36d4de 100644 --- a/internal/serviceroute/dataplane.go +++ b/internal/serviceroute/dataplane.go @@ -221,7 +221,10 @@ func (p *EBPFRouteProgrammer) Apply(intent RouteIntent) error { if err := p.ensureOpen(); err != nil { return err } - entries, err := p.entries(intent) + // An interface that is not on this node yet, or not registered yet, holds + // the intent with nothing installed: repairDesired installs it when the + // interface appears. + entries, err := p.recoverableEntries(intent) if err != nil { return err } @@ -253,6 +256,27 @@ func (p *EBPFRouteProgrammer) Apply(intent RouteIntent) error { return nil } +// Holding reports whether any of intents is held with nothing installed +// because its interface is missing or has no attachment identity yet. The CNI +// writes an interface's identity after the interface appears, with no +// interface event of its own, so a caller rechecks a policy that holds one. +func (p *EBPFRouteProgrammer) Holding(intents []RouteIntent) bool { + p.mu.Lock() + defer p.mu.Unlock() + for _, intent := range intents { + key := intentKey(intent) + if len(p.desiredIntents[key]) == 0 { + continue + } + for _, applied := range p.appliedRefs[key] { + if len(applied) == 0 { + return true + } + } + } + return false +} + func (p *EBPFRouteProgrammer) Remove(intent RouteIntent) error { p.mu.Lock() defer p.mu.Unlock() @@ -609,12 +633,14 @@ func (p *EBPFRouteProgrammer) closeHandles() { p.inheritedGate = false } -// Missing links withdraw their entries while preserving intent for recreation. -// Other lookup errors cannot prove that an interface has disappeared. +// Missing links, and links with no attachment identity, withdraw their entries +// while preserving intent for recreation. Other lookup errors cannot prove that +// an interface has disappeared. func (p *EBPFRouteProgrammer) recoverableEntries(intent RouteIntent) ([]appliedEntry, error) { entries, err := p.entries(intent) var missing netlink.LinkNotFoundError - if errors.As(err, &missing) || errors.Is(err, errNoProducerInterface) { + if errors.As(err, &missing) || errors.Is(err, errNoProducerInterface) || + errors.Is(err, serviceroutemap.ErrNoAttachmentIdentity) { return []appliedEntry{}, nil } return entries, err diff --git a/internal/serviceroute/dataplane_replacement_test.go b/internal/serviceroute/dataplane_replacement_test.go index ed47872b..07608da7 100644 --- a/internal/serviceroute/dataplane_replacement_test.go +++ b/internal/serviceroute/dataplane_replacement_test.go @@ -7,7 +7,6 @@ package serviceroute import ( - "errors" "fmt" "reflect" "testing" @@ -189,16 +188,21 @@ func TestLocalSetWithNoProducerInterfaceIsWithdrawn(t *testing.T) { } delete(f.links, producerADevice) delete(f.links, producerBDevice) - err := f.programmer.Apply(localSetIntent(producer(producerADevice, true), producer(producerBDevice, false))) - if !errors.As(err, &netlink.LinkNotFoundError{}) { - t.Fatalf("Apply with no producer interface = %v, want a missing link error", err) + next := localSetIntent(producer(producerADevice, true), producer(producerBDevice, false)) + if err := f.programmer.Apply(next); err != nil { + t.Fatalf("Apply with no producer interface = %v, want the route held", err) + } + if !f.programmer.Holding([]RouteIntent{next}) { + t.Fatal("Holding = false with every producer gone") + } + if err := f.programmer.Remove(intent); err != nil { + t.Fatal(err) } - f.repair(t, intent) if len(f.routes.rows) != 0 || len(f.set.rows) != 0 { t.Fatalf("routes=%d sets=%d with every producer gone, want the route withdrawn", len(f.routes.rows), len(f.set.rows)) } - if len(f.programmer.desiredIntents[intentKey(intent)]) != 1 { + if len(f.programmer.desiredIntents[intentKey(next)]) != 1 { t.Fatal("lost the intent to restore when a producer returns") } } diff --git a/internal/serviceroute/dataplane_test.go b/internal/serviceroute/dataplane_test.go index 4fd16fd4..b107d395 100644 --- a/internal/serviceroute/dataplane_test.go +++ b/internal/serviceroute/dataplane_test.go @@ -1035,3 +1035,73 @@ func TestInterfaceChurnKeepsUnrelatedRoutesServing(t *testing.T) { } observe() } + +// A consumer whose interface is not on the node yet, or has no attachment +// identity yet, is held with nothing installed instead of failing the policy, +// and the next repair installs it once the interface is ready. +func TestApplyHoldsConsumerUntilItsInterfaceIsReady(t *testing.T) { + for _, tc := range []struct { + name string + block func(programmer *EBPFRouteProgrammer, ready *bool) + }{ + {"interface missing", func(programmer *EBPFRouteProgrammer, ready *bool) { + programmer.targetIndexFn = func(name string) (uint32, error) { + if name == testConsumerDeviceName && !*ready { + return 0, netlink.LinkNotFoundError{} + } + if name == testConsumerDeviceName { + return 10, nil + } + return 20, nil + } + }}, + {"identity missing", func(programmer *EBPFRouteProgrammer, ready *bool) { + programmer.targetIndexFn = func(name string) (uint32, error) { + if name == testConsumerDeviceName && !*ready { + return 77, nil + } + if name == testConsumerDeviceName { + return 10, nil + } + return 20, nil + } + }}, + } { + t.Run(tc.name, func(t *testing.T) { + programmer := testRouteProgrammer(&recordingTable{}, &recordingTable{}, &recordingTable{}) + programmer.mapIDs = mapSetIDs{1, 2, 3, 4, 5, 6} + programmer.mapIDsValid = true + programmer.readMapIDsFn = func(string) (mapSetIDs, error) { return programmer.mapIDs, nil } + ready := false + tc.block(programmer, &ready) + intent := localTestIntent() + if err := programmer.Apply(intent); err != nil { + t.Fatalf("Apply = %v, want the consumer held", err) + } + if len(programmer.routeRefs) != 0 { + t.Fatal("held consumer installed routes") + } + if len(programmer.desiredIntents[intentKey(intent)]) != 1 { + t.Fatal("held consumer was not kept for repair") + } + if !programmer.Holding([]RouteIntent{intent}) { + t.Fatal("Holding = false while the consumer is held") + } + other := intent + other.Attachment.Name = "another-consumer" + if programmer.Holding([]RouteIntent{other}) { + t.Fatal("Holding = true for an intent of another policy") + } + ready = true + if err := programmer.Initialize(); err != nil { + t.Fatalf("repair once ready: %v", err) + } + if len(programmer.routeRefs) == 0 { + t.Fatal("ready consumer was not installed") + } + if programmer.Holding([]RouteIntent{intent}) { + t.Fatal("Holding = true after the consumer was installed") + } + }) + } +}