diff --git a/docs/router/architecture.md b/docs/router/architecture.md index 3c8baffe..b0dbfc4a 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 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. - **Deleting a `ServiceVIPBinding` does not need its VRF on the node.** A binding's two `vip_xlat_table` rows are keyed on its VRF's uSID Argument. `ServiceVIPBindingReconciler` reads it from the `BGPVRFInstance` its `vpcRef` names on this node, and that is gone once the VPC leaves the node. Before #750, deletion re-derived it, failed, held the finalizer forever and left both rows in the map. Teardown now finds the rows by value with `vipxlatmap.UnregisterBinding`: since #764 each key carries the VIP or backend address, and since #799 the ingress key also carries the backend's slot, so the binding's ingress row (its VIP, port and slot, rewriting to its backend) and egress row (its backend and port, rewriting to its VIP) are found exactly, under any Argument. Each row is removed only while it still holds the binding's values, so a row another binding has since written stays. When the VRF context still resolves, `UnregisterBindingAt` applies the same checks at that location too. Neither pass deletes a row without first reading it, and a process-wide lock keeps a bind from landing between the read and the delete. A value check cannot tell two bindings with identical values apart, so while such a twin is live on the node, deleting one removes nothing. Rows exist only for an IPv6 backend, so an IPv4 or unparseable `backendAddress` skips both passes instead of blocking deletion. Only a failed kernel delete keeps the finalizer. Rows left behind some other way, by a finalizer cleared by hand, a crash or a manual write, are removed by the periodic sweep below; to remove one sooner, use `galactic-router vip xlat list` and `vip xlat remove --protocol --vip --port --backend --backend-port` (`--slot` defaults to the backend's own), run in that node's `galactic-router` pod (`kubectl exec -- /galactic-router vip xlat list`). diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index ec2e5463..4680e06b 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -30,6 +30,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/source" cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" @@ -64,6 +65,12 @@ type ServiceRoutePolicyReconciler struct { mu sync.Mutex Applied map[types.NamespacedName]map[serviceroute.IntentSlot]serviceroute.RouteIntent + + // conflicted is the conflict set the last reconcile computed, and + // requeue carries the policies that entered or left it since + // (noteConflicts). + conflicted map[types.NamespacedName]bool + requeue chan event.GenericEvent } // compile compiles policy for node with this router's options. @@ -228,6 +235,7 @@ func (r *ServiceRoutePolicyReconciler) reconcilePolicy( *errorReason = serviceroute.ReconcileConflictLookup return ctrl.Result{}, errors.Join(conflictErr, r.removePolicyLocked(req.NamespacedName)) } + r.noteConflicts(req.NamespacedName, conflicts) if conflicts[req.NamespacedName] { var errs []error for key := range conflicts { @@ -346,9 +354,12 @@ func (r *ServiceRoutePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error })); err != nil { return fmt.Errorf("add initial service route policy map sync: %w", err) } + r.requeue = make(chan event.GenericEvent, 1024) return ctrl.NewControllerManagedBy(mgr). For(&networkv1alpha1.ServiceRoutePolicy{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). - Watches(&networkv1alpha1.ServiceRoutePolicy{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies), + // A policy another policy's reconcile moved into or out of conflict. + WatchesRawSource(source.Channel(r.requeue, &handler.EnqueueRequestForObject{})). + Watches(&networkv1alpha1.ServiceRoutePolicy{}, handler.EnqueueRequestsFromMapFunc(r.policyEventPolicies), builder.WithPredicates(predicate.GenerationChangedPredicate{})). // A renewal of producer entries that are still current changes no // route; expiry is handled by requeueing. @@ -360,8 +371,10 @@ func (r *ServiceRoutePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error builder.WithPredicates(predicate.Funcs{UpdateFunc: func(e event.UpdateEvent) bool { return !producerReportsOnlyRenewed(e.ObjectOld, e.ObjectNew, time.Now()) }})). - Watches(&cloudv1alpha1.VPCAttachment{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies)). - Watches(&cloudv1alpha1.VPC{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies)). + // An attachment or VPC event requeues only the policies it can change + // (serviceroute_requeue.go). + Watches(&cloudv1alpha1.VPCAttachment{}, r.attachmentEventPolicies()). + Watches(&cloudv1alpha1.VPC{}, handler.EnqueueRequestsFromMapFunc(r.vpcPolicies)). // Only the spec of a BGPRouter or BGPVRFInstance feeds an attachment's // SID. Their status, and the router's config-hash annotation, change // on nearly every BGPRouter reconcile, and each change would requeue @@ -635,14 +648,6 @@ func (r *ServiceRoutePolicyReconciler) withProducerReports(ctx context.Context, return serviceplan.WithProducerReports(endpoint, reports.Items), nil } -func (r *ServiceRoutePolicyReconciler) endpointPolicies(ctx context.Context, obj client.Object) []ctrl.Request { - return r.allPolicies(ctx) -} - -func (r *ServiceRoutePolicyReconciler) attachmentPolicies(ctx context.Context, _ client.Object) []ctrl.Request { - return r.allPolicies(ctx) -} - // routingDependencyChanged passes the BGPRouter and BGPVRFInstance events that // can change an attachment's SID: creation, deletion, and a spec change. var routingDependencyChanged = predicate.GenerationChangedPredicate{} diff --git a/internal/controller/serviceroute_requeue.go b/internal/controller/serviceroute_requeue.go new file mode 100644 index 00000000..6b019792 --- /dev/null +++ b/internal/controller/serviceroute_requeue.go @@ -0,0 +1,235 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package controller + +import ( + "context" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/labels" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" +) + +// requestQueue is the controller's work queue. +type requestQueue = workqueue.TypedRateLimitingInterface[reconcile.Request] + +// The service route controller has one worker, and every reconcile reads every +// attachment. Requeueing every policy for each attachment, endpoint or VPC +// event in the cell keeps its queue full, so a new consumer waits a whole +// pass over every policy before its route is installed. These mappers requeue +// only the policies an event can change. + +// attachmentEventPolicies requeues, for a VPCAttachment event, the policies +// that select the attachment as it was or as it is: as a consumer, or as a +// producer of their endpoint. An attachment leaving a selector requeues the +// policy it left, so that policy withdraws its route. Policies that conflict +// over a consumer both select it, so both are requeued for a consumer's +// event; a conflict that a producer, endpoint or VPC event creates or clears +// requeues the other policies through noteConflicts. +func (r *ServiceRoutePolicyReconciler) attachmentEventPolicies() handler.EventHandler { + enqueue := func(ctx context.Context, q requestQueue, objs ...client.Object) { + attachments := make([]*cloudv1alpha1.VPCAttachment, 0, len(objs)) + for _, obj := range objs { + if a, ok := obj.(*cloudv1alpha1.VPCAttachment); ok && a != nil { + attachments = append(attachments, a) + } + } + for _, req := range r.attachmentPolicies(ctx, attachments...) { + q.Add(req) + } + } + return handler.Funcs{ + CreateFunc: func(ctx context.Context, e event.CreateEvent, q requestQueue) { + enqueue(ctx, q, e.Object) + }, + UpdateFunc: func(ctx context.Context, e event.UpdateEvent, q requestQueue) { + enqueue(ctx, q, e.ObjectOld, e.ObjectNew) + }, + DeleteFunc: func(ctx context.Context, e event.DeleteEvent, q requestQueue) { + enqueue(ctx, q, e.Object) + }, + GenericFunc: func(ctx context.Context, e event.GenericEvent, q requestQueue) { + enqueue(ctx, q, e.Object) + }, + } +} + +// attachmentPolicies returns the policies that select any of attachments as +// a consumer or as a producer of their endpoint. A policy whose selector does +// not parse, or whose endpoint is not in the cache, is returned too: its +// reconcile reports the problem, and requeueing it costs one reconcile. +func (r *ServiceRoutePolicyReconciler) attachmentPolicies(ctx context.Context, + attachments ...*cloudv1alpha1.VPCAttachment, +) []ctrl.Request { + if len(attachments) == 0 { + return nil + } + policies := &networkv1alpha1.ServiceRoutePolicyList{} + if err := r.List(ctx, policies); err != nil { + log.FromContext(ctx).Error(err, "list service route policies for an attachment event") + return nil + } + endpoints := map[types.NamespacedName]*networkv1alpha1.ServiceEndpoint{} + endpointFor := func(policy *networkv1alpha1.ServiceRoutePolicy) (*networkv1alpha1.ServiceEndpoint, bool) { + key := types.NamespacedName{Namespace: policy.Namespace, Name: policy.Spec.ServiceRef.Name} + if endpoint, ok := endpoints[key]; ok { + return endpoint, endpoint != nil + } + endpoint := &networkv1alpha1.ServiceEndpoint{} + if err := r.Get(ctx, key, endpoint); err != nil { + endpoints[key] = nil + return nil, false + } + endpoints[key] = endpoint + return endpoint, true + } + var requests []ctrl.Request + for i := range policies.Items { + policy := &policies.Items[i] + endpoint, found := endpointFor(policy) + if !found || policySelectsAny(policy, endpoint, attachments) { + requests = append(requests, policyRequest(policy)) + } + } + return requests +} + +// policySelectsAny reports whether policy selects any of attachments as a +// consumer, or endpoint selects any of them as a producer. It errs toward +// true: a selector that does not parse matches. +func policySelectsAny(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, + attachments []*cloudv1alpha1.VPCAttachment, +) bool { + consumers, consumerErr := metav1.LabelSelectorAsSelector(&policy.Spec.AttachmentSelector) + var producers labels.Selector + var producerErr error + if endpoint.Spec.AttachmentSelector != nil { + producers, producerErr = metav1.LabelSelectorAsSelector(endpoint.Spec.AttachmentSelector) + } + if consumerErr != nil || producerErr != nil { + return true + } + for _, a := range attachments { + set := labels.Set(a.Labels) + if consumers.Matches(set) && consumerInPolicyVPC(policy, a) { + return true + } + if ref := endpoint.Spec.AttachmentRef; ref != nil && ref.Namespace == a.Namespace && ref.Name == a.Name { + return true + } + if producers != nil && producers.Matches(set) { + return true + } + } + return false +} + +// consumerInPolicyVPC reports whether a can be one of policy's consumers by +// its consumer VPC reference. The VPC identity in a's status is not compared: +// an attachment whose identity changes still requeues the policy that served +// it. +func consumerInPolicyVPC(policy *networkv1alpha1.ServiceRoutePolicy, a *cloudv1alpha1.VPCAttachment) bool { + ref := policy.Spec.ConsumerVPCRef + return ref == nil || (a.Namespace == policy.Namespace && a.Spec.VPC.Name == ref.Name) +} + +// endpointPolicies requeues the policies that route to the ServiceEndpoint +// obj, or to the endpoint whose producers the ServiceEndpointProducerReport +// obj describes. +func (r *ServiceRoutePolicyReconciler) endpointPolicies(ctx context.Context, obj client.Object) []ctrl.Request { + name := obj.GetName() + if report, ok := obj.(*networkv1alpha1.ServiceEndpointProducerReport); ok { + name = report.Spec.EndpointName + } + policies := &networkv1alpha1.ServiceRoutePolicyList{} + if err := r.List(ctx, policies, client.InNamespace(obj.GetNamespace())); err != nil { + log.FromContext(ctx).Error(err, "list service route policies for an endpoint event") + return nil + } + var requests []ctrl.Request + for i := range policies.Items { + if policies.Items[i].Spec.ServiceRef.Name == name { + requests = append(requests, policyRequest(&policies.Items[i])) + } + } + return requests +} + +// vpcPolicies requeues the policies whose consumer VPC reference names the +// VPC obj, the only policies that read a VPC. +func (r *ServiceRoutePolicyReconciler) vpcPolicies(ctx context.Context, obj client.Object) []ctrl.Request { + policies := &networkv1alpha1.ServiceRoutePolicyList{} + if err := r.List(ctx, policies, client.InNamespace(obj.GetNamespace())); err != nil { + log.FromContext(ctx).Error(err, "list service route policies for a VPC event") + return nil + } + var requests []ctrl.Request + for i := range policies.Items { + if ref := policies.Items[i].Spec.ConsumerVPCRef; ref != nil && ref.Name == obj.GetName() { + requests = append(requests, policyRequest(&policies.Items[i])) + } + } + return requests +} + +// policyEventPolicies requeues every policy for a policy's create, delete or +// spec change: whether two policies conflict over a consumer is decided across +// all of them. +func (r *ServiceRoutePolicyReconciler) policyEventPolicies(ctx context.Context, _ client.Object) []ctrl.Request { + return r.allPolicies(ctx) +} + +func policyRequest(policy *networkv1alpha1.ServiceRoutePolicy) ctrl.Request { + return ctrl.Request{NamespacedName: types.NamespacedName{Namespace: policy.Namespace, Name: policy.Name}} +} + +// noteConflicts requeues, apart from self, every policy that entered or left +// the conflict set since the last reconcile. A conflict can start or end +// through an event that requeues only one of the policies involved, such as +// its producer becoming ready, and the reconcile that sees it withdraws or +// would restore the others' routes. The caller holds r.mu. +func (r *ServiceRoutePolicyReconciler) noteConflicts(self types.NamespacedName, + conflicts map[types.NamespacedName]bool, +) { + previous := r.conflicted + r.conflicted = make(map[types.NamespacedName]bool, len(conflicts)) + for key, conflicted := range conflicts { + if conflicted { + r.conflicted[key] = true + } + } + changed := func(key types.NamespacedName) { + if key == self || r.requeue == nil { + return + } + select { + case r.requeue <- event.GenericEvent{Object: &networkv1alpha1.ServiceRoutePolicy{ + ObjectMeta: metav1.ObjectMeta{Namespace: key.Namespace, Name: key.Name}, + }}: + default: + // Full: the policy's own periodic requeue still reaches it. + } + } + for key := range r.conflicted { + if !previous[key] { + changed(key) + } + } + for key := range previous { + if !r.conflicted[key] { + changed(key) + } + } +} diff --git a/internal/controller/serviceroute_requeue_test.go b/internal/controller/serviceroute_requeue_test.go new file mode 100644 index 00000000..dfb9cbac --- /dev/null +++ b/internal/controller/serviceroute_requeue_test.go @@ -0,0 +1,249 @@ +// Copyright 2026 Datum Cloud, Inc. +// +// SPDX-License-Identifier: AGPL-3.0-or-later + +package controller + +import ( + "context" + "slices" + "testing" + + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + utilruntime "k8s.io/apimachinery/pkg/util/runtime" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" + networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" +) + +const ( + requeueNS = "tenant" + requeueGuest = "guest" + requeueAPI = "api" + requeueDNSA = "rq-dns-a" + requeueDNSB = "rq-dns-b" + requeueVPCA = "rq-vpc-a" + requeueVPCB = "rq-vpc-b" + requeueRole = "role" + requeueApp = "app" +) + +func requeuePolicy(name, endpoint string, selector map[string]string, + vpc *networkv1alpha1.ServiceRouteVPCReference, +) *networkv1alpha1.ServiceRoutePolicy { + return &networkv1alpha1.ServiceRoutePolicy{ + ObjectMeta: metav1.ObjectMeta{Namespace: requeueNS, Name: name}, + Spec: networkv1alpha1.ServiceRoutePolicySpec{ + ServiceRef: networkv1alpha1.ServiceEndpointReference{Name: endpoint}, + AttachmentSelector: metav1.LabelSelector{MatchLabels: selector}, + ConsumerVPCRef: vpc, + }, + } +} + +func requeueEndpoint(name string, selector map[string]string, + ref *networkv1alpha1.ServiceEndpointAttachmentReference, +) *networkv1alpha1.ServiceEndpoint { + e := &networkv1alpha1.ServiceEndpoint{ObjectMeta: metav1.ObjectMeta{Namespace: requeueNS, Name: name}} + if ref != nil { + e.Spec.AttachmentRef = ref + } else { + e.Spec.AttachmentSelector = &metav1.LabelSelector{MatchLabels: selector} + } + return e +} + +func requeueAttachment(name, vpc string, l map[string]string) *cloudv1alpha1.VPCAttachment { + a := &cloudv1alpha1.VPCAttachment{ObjectMeta: metav1.ObjectMeta{Namespace: requeueNS, Name: name, Labels: l}} + a.Spec.VPC.Name = vpc + return a +} + +func newRequeueReconciler(t *testing.T, objs ...client.Object) *ServiceRoutePolicyReconciler { + t.Helper() + scheme := runtime.NewScheme() + utilruntime.Must(networkv1alpha1.AddToScheme(scheme)) + utilruntime.Must(cloudv1alpha1.AddToScheme(scheme)) + return &ServiceRoutePolicyReconciler{ + Client: fake.NewClientBuilder().WithScheme(scheme).WithObjects(objs...).Build(), + } +} + +func requestNames(reqs []ctrl.Request) []string { + names := make([]string, 0, len(reqs)) + for _, r := range reqs { + names = append(names, r.Name) + } + slices.Sort(names) + return slices.Compact(names) +} + +// requeueObjects is a small cell: a DNS policy whose consumers are every +// guest in vpc-a, served by a selector endpoint; a policy for vpc-b only; and +// a policy whose endpoint names its one producer. +func requeueObjects() []client.Object { + return []client.Object{ + requeuePolicy(requeueDNSA, "dns", map[string]string{requeueRole: requeueGuest}, + &networkv1alpha1.ServiceRouteVPCReference{Name: requeueVPCA, UID: "a"}), + requeuePolicy(requeueDNSB, "dns", map[string]string{requeueRole: requeueGuest}, + &networkv1alpha1.ServiceRouteVPCReference{Name: requeueVPCB, UID: "b"}), + requeuePolicy("db", "db", map[string]string{requeueApp: requeueAPI}, nil), + requeueEndpoint("dns", map[string]string{requeueRole: "dns-member"}, nil), + requeueEndpoint("db", nil, &networkv1alpha1.ServiceEndpointAttachmentReference{ + Namespace: requeueNS, Name: "db-0", + }), + } +} + +func TestAttachmentPoliciesRequeuesOnlyPoliciesThatSelectIt(t *testing.T) { + r := newRequeueReconciler(t, requeueObjects()...) + ctx := context.Background() + for _, tc := range []struct { + name string + attachment *cloudv1alpha1.VPCAttachment + want []string + }{ + {"guest in vpc-a", requeueAttachment("guest-1", requeueVPCA, + map[string]string{requeueRole: requeueGuest}), []string{requeueDNSA}}, + {"guest in vpc-b", requeueAttachment("guest-2", requeueVPCB, + map[string]string{requeueRole: requeueGuest}), []string{requeueDNSB}}, + {"DNS member, a producer for both", requeueAttachment("member-0", "infra", + map[string]string{requeueRole: "dns-member"}), []string{requeueDNSA, requeueDNSB}}, + {"producer named by attachmentRef", requeueAttachment("db-0", "infra", nil), []string{"db"}}, + {"consumer of the policy without a VPC reference", requeueAttachment("api-0", "vpc-c", + map[string]string{requeueApp: requeueAPI}), []string{"db"}}, + {"unrelated attachment", requeueAttachment("other", requeueVPCA, map[string]string{requeueRole: "batch"}), nil}, + } { + t.Run(tc.name, func(t *testing.T) { + got := requestNames(r.attachmentPolicies(ctx, tc.attachment)) + if !slices.Equal(got, tc.want) { + t.Fatalf("requeued %v, want %v", got, tc.want) + } + }) + } +} + +// An update requeues the policy the attachment left as well as the one it +// joined, so the old policy withdraws its route. +func TestAttachmentEventRequeuesOldAndNewSelectors(t *testing.T) { + r := newRequeueReconciler(t, requeueObjects()...) + old := requeueAttachment("guest-1", requeueVPCA, map[string]string{requeueRole: requeueGuest}) + moved := requeueAttachment("guest-1", requeueVPCA, map[string]string{requeueApp: requeueAPI}) + q := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]()) + defer q.ShutDown() + + r.attachmentEventPolicies().Update(context.Background(), event.UpdateEvent{ObjectOld: old, ObjectNew: moved}, q) + + var got []string + for q.Len() > 0 { + req, _ := q.Get() + got = append(got, req.Name) + q.Done(req) + } + slices.Sort(got) + if want := []string{"db", requeueDNSA}; !slices.Equal(got, want) { + t.Fatalf("requeued %v, want %v", got, want) + } +} + +// A policy whose endpoint is missing, or whose selector does not parse, is +// requeued for any attachment event so its reconcile can report it. +func TestAttachmentPoliciesKeepsPoliciesItCannotJudge(t *testing.T) { + broken := requeuePolicy("broken", "dns", nil, nil) + broken.Spec.AttachmentSelector = metav1.LabelSelector{MatchExpressions: []metav1.LabelSelectorRequirement{{ + Key: "role", Operator: "NotAnOperator", + }}} + orphan := requeuePolicy("orphan", "missing", map[string]string{requeueRole: requeueGuest}, nil) + r := newRequeueReconciler(t, append(requeueObjects(), broken, orphan)...) + + got := requestNames(r.attachmentPolicies(context.Background(), + requeueAttachment("other", requeueVPCA, map[string]string{requeueRole: "batch"}))) + if want := []string{"broken", "orphan"}; !slices.Equal(got, want) { + t.Fatalf("requeued %v, want %v", got, want) + } +} + +func TestEndpointAndReportEventsRequeueTheirPolicies(t *testing.T) { + r := newRequeueReconciler(t, requeueObjects()...) + ctx := context.Background() + db := requeueEndpoint("db", nil, + &networkv1alpha1.ServiceEndpointAttachmentReference{Namespace: requeueNS, Name: "db-0"}) + if got, want := requestNames(r.endpointPolicies(ctx, db)), []string{"db"}; !slices.Equal(got, want) { + t.Fatalf("endpoint db requeued %v, want %v", got, want) + } + report := &networkv1alpha1.ServiceEndpointProducerReport{ + ObjectMeta: metav1.ObjectMeta{Namespace: requeueNS, Name: "dns-node-a"}, + Spec: networkv1alpha1.ServiceEndpointProducerReportSpec{EndpointName: "dns", Node: "rq-node"}, + } + got, want := requestNames(r.endpointPolicies(ctx, report)), []string{requeueDNSA, requeueDNSB} + if !slices.Equal(got, want) { + t.Fatalf("report for dns requeued %v, want %v", got, want) + } + other := &networkv1alpha1.ServiceEndpoint{ObjectMeta: metav1.ObjectMeta{Namespace: "rq-other-ns", Name: "dns"}} + if got := r.endpointPolicies(ctx, other); len(got) != 0 { + t.Fatalf("an endpoint in another namespace requeued %v", requestNames(got)) + } +} + +func TestVPCEventRequeuesPoliciesThatReferenceIt(t *testing.T) { + r := newRequeueReconciler(t, requeueObjects()...) + vpc := &cloudv1alpha1.VPC{ObjectMeta: metav1.ObjectMeta{Namespace: requeueNS, Name: requeueVPCB}} + got, want := requestNames(r.vpcPolicies(context.Background(), vpc)), []string{requeueDNSB} + if !slices.Equal(got, want) { + t.Fatalf("requeued %v, want %v", got, want) + } +} + +// A reconcile that sees another policy enter or leave the conflict set +// requeues it, since that policy's own inputs may not have changed. +func TestNoteConflictsRequeuesPoliciesThatEnteredOrLeft(t *testing.T) { + key := func(name string) types.NamespacedName { return types.NamespacedName{Namespace: requeueNS, Name: name} } + r := &ServiceRoutePolicyReconciler{requeue: make(chan event.GenericEvent, 8)} + drain := func() []string { + var names []string + for len(r.requeue) > 0 { + names = append(names, (<-r.requeue).Object.GetName()) + } + slices.Sort(names) + return names + } + + r.noteConflicts(key("b"), map[types.NamespacedName]bool{key("a"): true, key("b"): true}) + if got, want := drain(), []string{"a"}; !slices.Equal(got, want) { + t.Fatalf("entering conflict requeued %v, want %v", got, want) + } + r.noteConflicts(key("b"), map[types.NamespacedName]bool{key("a"): true, key("b"): true}) + if got := drain(); len(got) != 0 { + t.Fatalf("an unchanged conflict set requeued %v", got) + } + r.noteConflicts(key("b"), map[types.NamespacedName]bool{}) + if got, want := drain(), []string{"a"}; !slices.Equal(got, want) { + t.Fatalf("leaving conflict requeued %v, want %v", got, want) + } +} + +// A delete event requeues the policies the deleted attachment was selected by. +func TestAttachmentEventDeleteRequeuesItsPolicies(t *testing.T) { + r := newRequeueReconciler(t, requeueObjects()...) + q := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[reconcile.Request]()) + defer q.ShutDown() + gone := requeueAttachment("guest-1", requeueVPCA, map[string]string{requeueRole: requeueGuest}) + + r.attachmentEventPolicies().Delete(context.Background(), event.DeleteEvent{Object: gone}, q) + + if q.Len() != 1 { + t.Fatalf("queue holds %d requests, want 1", q.Len()) + } + req, _ := q.Get() + if req.Name != requeueDNSA { + t.Fatalf("requeued %s, want %s", req.Name, requeueDNSA) + } +}