From eb6434df667ecb33a3d6ef37dac9fb848909d1cc Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 12:41:34 -0400 Subject: [PATCH 1/2] fix: Expire producer health reports a node stops renewing Each node writes the health of the Spread producers it hosts into the endpoint's status.producers, and every node routes on those entries. An entry stayed until its own node replaced it, so a node that went down, or whose galactic-router stopped, left its producers reported as they last were. A remote consumer kept sending new flows to a lost node's producers while their attachments still read ready, and a producer reported Unhealthy just before its router stopped stayed drained after it recovered. Each entry now carries renewTime. Every node renews its entries every 15 seconds, even when no decision changed, and skips the write while its entries are unchanged and younger than that. An entry is current for 60 seconds after its renewTime; after that, or with no renewTime, it is expired. When the planner and the programming plan pick producers, they read an expired entry this way: - It never drains a producer in a node's local set. - A remote consumer goes to the first producer currently reported Healthy or Unknown, or not reported at all. Failing that, it goes to the first producer whose entry expired, and then to the first producer, as it already did when every producer is Unhealthy. ServiceRoutePolicyReconciler requeues for the next entry expiry, so a node with no intents yet picks up a remote producer that moves onto it. It ignores endpoint updates that only renew entries still current, so the periodic renewal does not reconcile every policy on every node. A router without renewal writes no renewTime, so during a rollout, upgraded nodes rank the producers of nodes not yet upgraded after currently reported ones. A rolled-back router ignores the field. The private services guide documents the timing, the ordering and this rollout behavior. Resolves #1099 --- api/network/v1alpha1/serviceroute_types.go | 23 +++ api/network/v1alpha1/zz_generated.deepcopy.go | 4 + ...lactic.datumapis.com_serviceendpoints.yaml | 12 +- docs/api/network-gateway.md | 3 +- docs/private-services/README.md | 35 ++++- docs/router/architecture.md | 2 +- .../serviceproducerhealth_controller.go | 55 ++++++-- .../serviceproducerhealth_controller_test.go | 72 +++++++++- .../controller/serviceroute_controller.go | 53 ++++++- .../controller/serviceroute_spread_test.go | 45 ++++++ internal/serviceroute/planner.go | 97 +++++++++---- internal/serviceroute/planner_spread_test.go | 133 +++++++++++++++++- internal/serviceroute/programming.go | 5 +- 13 files changed, 483 insertions(+), 56 deletions(-) diff --git a/api/network/v1alpha1/serviceroute_types.go b/api/network/v1alpha1/serviceroute_types.go index 911bc129..5750a07b 100644 --- a/api/network/v1alpha1/serviceroute_types.go +++ b/api/network/v1alpha1/serviceroute_types.go @@ -1,6 +1,8 @@ package v1alpha1 import ( + "time" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" ) @@ -128,6 +130,18 @@ const ( // endpoint that a node can hold. const ServiceEndpointMaxProducersPerNode = 16 +const ( + // ServiceEndpointProducerReportRenewInterval is how often a node renews + // its entries in a Spread endpoint's status.producers while its decisions + // do not change. + ServiceEndpointProducerReportRenewInterval = 15 * time.Second + + // ServiceEndpointProducerReportLifetime is how long after its renewTime + // an entry in status.producers is current. Nodes route on an entry only + // while it is current. + ServiceEndpointProducerReportLifetime = 60 * time.Second +) + // ServiceEndpointHealthCheck configures the TCP health check of an endpoint's // producers. A probe opens a TCP connection to the producer attachment's first // IPv6 address, not the endpoint's address, on Port, through the producer's @@ -188,6 +202,9 @@ type ServiceEndpointStatus struct { // under its own field manager, and removes them when the producer stops // being selected. An entry whose attachmentUID no longer matches the // attachment of that name describes an earlier attachment and is ignored. + // Each node renews its entries every 15 seconds; an entry not renewed + // for 60 seconds is expired, because its node or its galactic-router has + // stopped, and no node routes on its state. // +optional // +listType=map // +listMapKey=namespace @@ -238,6 +255,12 @@ type ServiceEndpointProducerStatus struct { // LastTransitionTime is when State last changed. // +kubebuilder:validation:Required LastTransitionTime metav1.Time `json:"lastTransitionTime"` + + // RenewTime is when the producer's node last confirmed this entry. The + // entry is current for 60 seconds after it, and expired after that or + // when RenewTime is unset. + // +optional + RenewTime *metav1.Time `json:"renewTime,omitempty"` } // ServiceEndpointDeliveryMode defines the placement contract for a service. diff --git a/api/network/v1alpha1/zz_generated.deepcopy.go b/api/network/v1alpha1/zz_generated.deepcopy.go index 083ac983..c9cd9e9a 100644 --- a/api/network/v1alpha1/zz_generated.deepcopy.go +++ b/api/network/v1alpha1/zz_generated.deepcopy.go @@ -4279,6 +4279,10 @@ func (in *ServiceEndpointList) DeepCopyObject() runtime.Object { func (in *ServiceEndpointProducerStatus) DeepCopyInto(out *ServiceEndpointProducerStatus) { *out = *in in.LastTransitionTime.DeepCopyInto(&out.LastTransitionTime) + if in.RenewTime != nil { + in, out := &in.RenewTime, &out.RenewTime + *out = (*in).DeepCopy() + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ServiceEndpointProducerStatus. diff --git a/config/crd/galactic.datumapis.com_serviceendpoints.yaml b/config/crd/galactic.datumapis.com_serviceendpoints.yaml index 7abb61f7..380d7c90 100644 --- a/config/crd/galactic.datumapis.com_serviceendpoints.yaml +++ b/config/crd/galactic.datumapis.com_serviceendpoints.yaml @@ -3,7 +3,7 @@ apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: annotations: - controller-gen.kubebuilder.io/version: v0.21.0 + controller-gen.kubebuilder.io/version: v0.18.0 name: serviceendpoints.galactic.datumapis.com spec: group: galactic.datumapis.com @@ -276,6 +276,9 @@ spec: under its own field manager, and removes them when the producer stops being selected. An entry whose attachmentUID no longer matches the attachment of that name describes an earlier attachment and is ignored. + Each node renews its entries every 15 seconds; an entry not renewed + for 60 seconds is expired, because its node or its galactic-router has + stopped, and no node routes on its state. items: description: ServiceEndpointProducerStatus is one producer's health decision. @@ -317,6 +320,13 @@ spec: or ProbeUnavailable. maxLength: 64 type: string + renewTime: + description: |- + RenewTime is when the producer's node last confirmed this entry. The + entry is current for 60 seconds after it, and expired after that or + when RenewTime is unset. + format: date-time + type: string state: description: |- State is the decision. Healthy and Unknown producers receive new diff --git a/docs/api/network-gateway.md b/docs/api/network-gateway.md index 7f8993b9..7fdc3d27 100644 --- a/docs/api/network-gateway.md +++ b/docs/api/network-gateway.md @@ -687,6 +687,7 @@ _Appears in:_ | `reason` _string_ | Reason is ProbeSucceeded, ProbeFailed, Pending or ProbeUnavailable. | | MaxLength: 64
Required: \{\}
| | `message` _string_ | Message is the last probe error, when there is one. | | MaxLength: 1024
| | `lastTransitionTime` _[Time](https://kubernetes.io/docs/reference/generated/kubernetes-api/v/#time-v1-meta)_ | LastTransitionTime is when State last changed. | | Required: \{\}
| +| `renewTime` _[Time](https://kubernetes.io/docs/reference/generated/kubernetes-api/v/#time-v1-meta)_ | RenewTime is when the producer's node last confirmed this entry. The
entry is current for 60 seconds after it, and expired after that or
when RenewTime is unset. | | | #### ServiceEndpointReference @@ -744,7 +745,7 @@ _Appears in:_ | Field | Description | Default | Validation | | --- | --- | --- | --- | -| `producers` _[ServiceEndpointProducerStatus](#serviceendpointproducerstatus) array_ | Producers holds the health decision of every producer of a Spread
endpoint. Each node writes the entries of the producers on that node,
under its own field manager, and removes them when the producer stops
being selected. An entry whose attachmentUID no longer matches the
attachment of that name describes an earlier attachment and is ignored. | | MaxItems: 4096
| +| `producers` _[ServiceEndpointProducerStatus](#serviceendpointproducerstatus) array_ | Producers holds the health decision of every producer of a Spread
endpoint. Each node writes the entries of the producers on that node,
under its own field manager, and removes them when the producer stops
being selected. An entry whose attachmentUID no longer matches the
attachment of that name describes an earlier attachment and is ignored.
Each node renews its entries every 15 seconds; an entry not renewed
for 60 seconds is expired, because its node or its galactic-router has
stopped, and no node routes on its state. | | MaxItems: 4096
| #### ServiceRouteAuthorization diff --git a/docs/private-services/README.md b/docs/private-services/README.md index 1be0e152..8581f67f 100644 --- a/docs/private-services/README.md +++ b/docs/private-services/README.md @@ -84,12 +84,18 @@ TCP connection from `fe80::/10` on the Private path. The node publishes the decision for each of its producers in the endpoint's `status.producers`: ```console -$ kubectl get serviceendpoint resolver -o jsonpath='{range .status.producers[*]}{.name}{"\t"}{.node}{"\t"}{.state}{"\t"}{.reason}{"\n"}{end}' -resolver-a node-1 Healthy ProbeSucceeded -resolver-b node-1 Unhealthy ProbeFailed +$ kubectl get serviceendpoint resolver -o jsonpath='{range .status.producers[*]}{.name}{"\t"}{.node}{"\t"}{.state}{"\t"}{.reason}{"\t"}{.renewTime}{"\n"}{end}' +resolver-a node-1 Healthy ProbeSucceeded 2026-01-01T12:00:45Z +resolver-b node-1 Unhealthy ProbeFailed 2026-01-01T12:00:45Z ``` -What the decision does: +The node renews its entries every 15 seconds, whether or not a decision +changed, and sets each entry's `renewTime`. An entry is current for 60 seconds +after its `renewTime`. After that it is expired: its node is down, or its +`galactic-router` has stopped, and its `state` may no longer be true. An entry +with no `renewTime` is expired too. + +What a current decision does: | State | New flows | Existing TCP connections and replies | | ----------- | --------- | ------------------------------------ | @@ -97,6 +103,10 @@ What the decision does: | `Unknown` | yes | yes | | `Unhealthy` | no | yes | +An expired entry never drains a producer on its own node, so a producer +reported `Unhealthy` just before its node's router stopped takes new flows +again once the entry expires. + When every producer on a node is `Unhealthy`, all of them keep taking new flows, because an empty set would drop every request. A producer that crashes stops getting new flows within `intervalSeconds × unhealthyThreshold` plus @@ -111,9 +121,14 @@ longest request, then stop it. Its in-flight requests and open connections finish while new ones go to the other producers. When the consumer's node has no producer, `PreferNodeLocal` sends the consumer -to one remote producer, the first by namespace and name that its node has not -reported `Unhealthy`, through the SRv6 service tunnel as with `Single`. A -remote consumer therefore does not spread. While a health change reaches the +to one remote producer, through the SRv6 service tunnel as with `Single`. It is +the first producer by namespace and name whose node currently reports it +`Healthy` or `Unknown`, or has no entry for it. If there is none, it is the +first producer whose entry expired, and failing that the first producer. So +when a node goes down while its producers' attachments still read ready, +consumers on other nodes move off those producers within 60 seconds of the +node's last renewal, as long as another node currently reports a producer that +can take them. A remote consumer does not spread. While a health change reaches the consumer's node and the producer's node at different moments, the two can briefly disagree on that producer, and its requests are refused until they agree. @@ -127,6 +142,12 @@ the connection resets. Changing an endpoint between `Single` and `Spread` replaces its routes, which can drop a request in flight and moves existing TCP connections. +A `galactic-router` without report renewal writes no `renewTime`, so while +routers are being upgraded, consumers on upgraded nodes treat the entries of +nodes that are not yet upgraded as expired, and send remote consumers to +them only when no current entry offers a producer. This lasts until the +rollout finishes. A rolled-back router ignores `renewTime`. + Spread needs a `galactic-cni` whose datapath has the producer set map. A `galactic-router` that finds the map missing refuses the endpoint's Spread routes on that node, retrying, and keeps every other route working. Roll out diff --git a/docs/router/architecture.md b/docs/router/architecture.md index 9966f492..545c1212 100644 --- a/docs/router/architecture.md +++ b/docs/router/architecture.md @@ -120,7 +120,7 @@ there instead. - **A tenant VRF route that fails to install is retried, not dropped.** `processEVPNPath` writes a VRF-scoped path's route into `egress_route_table` through `internal/runtime/gobgp/vrfroutes.go`. `Register` resolves the SID's uplink and neighbor MAC at write time, which fails while an uplink is still coming up after a restart, and BGP never redelivers an unchanged best path, so before #676 such a route stayed missing until the path changed upstream. `vrfRoutes` keeps each failed route in a pending set and retries it whenever the main table changes (1s debounce) and every 30s, until it installs or its path is withdrawn; removing a VRF drops its pending routes. One mutex serializes every write to the map from this runtime, taken per route during a retry so an unresolvable neighbor (up to 2s) does not hold up the RIB watcher for the whole pass. `EgressRouteTable.Refresh` cannot cover this: it walks existing entries only. The pending count, and the plain-route reconciler's failing count, are exported as `galactic_router_route_install_failing{kind="vrf"|"plain"}`, and `GalacticRouterRoutesNotInstalled` fires when either stays above 0 for 10 minutes. - **`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 `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 entries of the endpoint's `status.producers` under its own field manager (`galactic-router/`), so nodes never overwrite each other's entries and an entry a node stops applying disappears. `ServiceRoutePolicyReconciler` reads them: for a Spread endpoint, `serviceroute.Compile` emits one `LocalSet` intent per consumer holding every producer on the consumer's node, `Unhealthy` ones marked draining unless all are. `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. +- **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 entries of the endpoint's `status.producers` under its own field manager (`galactic-router/`), so nodes never overwrite each other's entries and an entry a node stops applying disappears. 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 (`serviceroute.ProducerReportCurrent`). `ServiceRoutePolicyReconciler` requeues for the next expiry and ignores endpoint updates that only renew current entries. `ServiceRoutePolicyReconciler` reads them: for a Spread endpoint, `serviceroute.Compile` emits one `LocalSet` intent per consumer holding every producer on the consumer's node, `Unhealthy` ones marked draining unless all are. `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`). - **`vip_xlat_table` holds both VIP families in one layout.** The map (`struct vip_xlat_key`, 32 bytes: block, argument, protocol, direction, port, slot, 16-byte address; `struct vip_xlat_value`, 18 bytes: 16-byte address, port) is pinned by `galactic-cni` at `/sys/fs/bpf/galactic/vip_xlat_table` and reused across restarts and upgrades by `attach.Load`; `galactic-router`'s `ServiceVIPBindingReconciler` and the `vip xlat` CLI are its only writers. Its layout is fixed, since CNI ADD and binding time write the rows and a new layout would empty the map until every binding was written again. A binding writes two rows, by its VIP's family (every key also carries block, argument and protocol): diff --git a/internal/controller/serviceproducerhealth_controller.go b/internal/controller/serviceproducerhealth_controller.go index 1f71020f..eead45ba 100644 --- a/internal/controller/serviceproducerhealth_controller.go +++ b/internal/controller/serviceproducerhealth_controller.go @@ -48,7 +48,9 @@ import ( // node stops applying is removed. Each producer has one probe loop, started // from the decision already published for the same attachment and port, so a // restarted galactic-router changes no producer set. A producer with no IPv6 -// address cannot be probed and is published as Unknown. +// address cannot be probed and is published as Unknown. The node renews its +// entries every ServiceEndpointProducerReportRenewInterval, so other nodes stop +// routing on them once it is down or its galactic-router stops. type ServiceProducerHealthReconciler struct { client.Client NodeName string @@ -86,7 +88,7 @@ func (r *ServiceProducerHealthReconciler) Reconcile(ctx context.Context, req ctr if endpoint.Spec.Distribution != networkv1alpha1.ServiceEndpointDistributionSpread || endpoint.Spec.HealthCheck == nil || !endpoint.DeletionTimestamp.IsZero() { r.releaseExcept(req.NamespacedName, nil) - return ctrl.Result{}, r.publish(ctx, endpoint, nil) + return r.publish(ctx, endpoint, nil, time.Now()) } list := &cloudv1alpha1.VPCAttachmentList{} @@ -102,7 +104,7 @@ func (r *ServiceProducerHealthReconciler) Reconcile(ctx context.Context, req ctr // The policy controller reports an invalid endpoint; there is // nothing to probe. r.releaseExcept(req.NamespacedName, nil) - return ctrl.Result{}, r.publish(ctx, endpoint, nil) + return r.publish(ctx, endpoint, nil, time.Now()) } published := make(map[types.NamespacedName]networkv1alpha1.ServiceEndpointProducerStatus) @@ -141,7 +143,7 @@ func (r *ServiceProducerHealthReconciler) Reconcile(ctx context.Context, req ctr entries = append(entries, entry) } r.track(req.NamespacedName, wanted, entries) - return ctrl.Result{}, r.publish(ctx, endpoint, entries) + return r.publish(ctx, endpoint, entries, time.Now()) } // producerHealthCheckPort returns the port endpoint's healthCheck probes: its @@ -285,10 +287,12 @@ func (r *ServiceProducerHealthReconciler) recordStatesLocked(endpoint types.Name } // publish applies entries as this node's part of endpoint's status.producers, -// unless this node's published entries already equal them. +// each renewed at now, unless this node's published entries already equal +// them and were all renewed within the renew interval. It requeues for the +// next renewal while entries is not empty. func (r *ServiceProducerHealthReconciler) publish(ctx context.Context, endpoint *networkv1alpha1.ServiceEndpoint, - entries []networkv1alpha1.ServiceEndpointProducerStatus, -) error { + entries []networkv1alpha1.ServiceEndpointProducerStatus, now time.Time, +) (ctrl.Result, error) { var current []networkv1alpha1.ServiceEndpointProducerStatus for _, entry := range endpoint.Status.Producers { if entry.Node == r.NodeName { @@ -297,12 +301,25 @@ func (r *ServiceProducerHealthReconciler) publish(ctx context.Context, endpoint } sortProducerEntries(current) sortProducerEntries(entries) + renewIn := time.Duration(0) + if len(entries) != 0 { + renewIn = networkv1alpha1.ServiceEndpointProducerReportRenewInterval + } if producerEntriesEqual(current, entries) { - return nil + if len(entries) == 0 { + return ctrl.Result{}, nil + } + if age, ok := oldestRenewal(current, now); ok && age < renewIn { + return ctrl.Result{RequeueAfter: max(renewIn-age, time.Second)}, nil + } } if entries == nil { entries = []networkv1alpha1.ServiceEndpointProducerStatus{} } + renewed := metav1.NewTime(now.Truncate(time.Second)) + for i := range entries { + entries[i].RenewTime = &renewed + } body, err := json.Marshal(map[string]any{ "apiVersion": networkv1alpha1.GroupVersion.String(), "kind": "ServiceEndpoint", @@ -310,19 +327,33 @@ func (r *ServiceProducerHealthReconciler) publish(ctx context.Context, endpoint "status": map[string]any{"producers": entries}, }) if err != nil { - return err + return ctrl.Result{}, err } obj := &networkv1alpha1.ServiceEndpoint{} obj.Name, obj.Namespace = endpoint.Name, endpoint.Namespace err = r.Status().Patch(ctx, obj, client.RawPatch(types.ApplyPatchType, body), client.FieldOwner(nodeFieldOwner(r.NodeName)), client.ForceOwnership) if apierrors.IsNotFound(err) { - return nil + return ctrl.Result{}, nil } if err != nil { - return fmt.Errorf("apply status.producers of ServiceEndpoint %s/%s: %w", endpoint.Namespace, endpoint.Name, err) + return ctrl.Result{}, fmt.Errorf("apply status.producers of ServiceEndpoint %s/%s: %w", + endpoint.Namespace, endpoint.Name, err) + } + return ctrl.Result{RequeueAfter: renewIn}, nil +} + +// oldestRenewal returns the age at now of the least recently renewed entry, +// and false when an entry was never renewed. +func oldestRenewal(entries []networkv1alpha1.ServiceEndpointProducerStatus, now time.Time) (time.Duration, bool) { + var oldest time.Duration + for _, entry := range entries { + if entry.RenewTime == nil { + return 0, false + } + oldest = max(oldest, now.Sub(entry.RenewTime.Time)) } - return nil + return oldest, true } func sortProducerEntries(entries []networkv1alpha1.ServiceEndpointProducerStatus) { diff --git a/internal/controller/serviceproducerhealth_controller_test.go b/internal/controller/serviceproducerhealth_controller_test.go index e7fe23f5..8a69627a 100644 --- a/internal/controller/serviceproducerhealth_controller_test.go +++ b/internal/controller/serviceproducerhealth_controller_test.go @@ -32,6 +32,7 @@ const ( producerHealthNamespace = "dns" producerHealthNode = "node-a" producerHealthOtherNode = "node-b" + producerHealthClass = "dns" ) // ownerProber returns a decision per probe owner, Unknown/Pending for any @@ -147,7 +148,8 @@ func TestServiceProducerHealthReconcilerPublishesThisNodesProducers(t *testing.T endpoint := &networkv1alpha1.ServiceEndpoint{ ObjectMeta: metav1.ObjectMeta{Name: producerHealthEndpoint, Namespace: producerHealthNamespace}, Spec: networkv1alpha1.ServiceEndpointSpec{ - ServiceClass: "dns", Address: producerHealthAddress, Port: 53, Protocol: networkv1alpha1.NetworkServiceProtocolUDP, + ServiceClass: producerHealthClass, Address: producerHealthAddress, Port: 53, + Protocol: networkv1alpha1.NetworkServiceProtocolUDP, DeliveryMode: networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal, AttachmentSelector: &metav1.LabelSelector{ MatchLabels: map[string]string{producerHealthLabel: producerHealthEndpoint}, @@ -252,3 +254,71 @@ func TestServiceProducerHealthReconcilerPublishesThisNodesProducers(t *testing.T t.Errorf("probes after Single = %v, want none", prober.owners) } } + +func TestServiceProducerHealthReconcilerRenewsReports(t *testing.T) { + c := startProducerHealthAPI(t) + ctx := context.Background() + endpoint := &networkv1alpha1.ServiceEndpoint{ + ObjectMeta: metav1.ObjectMeta{Name: producerHealthEndpoint, Namespace: producerHealthNamespace}, + Spec: networkv1alpha1.ServiceEndpointSpec{ + ServiceClass: producerHealthClass, Address: producerHealthAddress, Port: 53, + Protocol: networkv1alpha1.NetworkServiceProtocolUDP, + DeliveryMode: networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal, + AttachmentSelector: &metav1.LabelSelector{ + MatchLabels: map[string]string{producerHealthLabel: producerHealthEndpoint}, + }, + Distribution: networkv1alpha1.ServiceEndpointDistributionSpread, + HealthCheck: &networkv1alpha1.ServiceEndpointHealthCheck{Port: 8053}, + }, + } + if err := c.Create(ctx, endpoint); err != nil { + t.Fatal(err) + } + createProducer(t, c, "resolver-a", producerHealthNode, "fd00:1::a/64") + key := client.ObjectKeyFromObject(endpoint) + r := &ServiceProducerHealthReconciler{Client: c, NodeName: producerHealthNode, Prober: newOwnerProber()} + interval := networkv1alpha1.ServiceEndpointProducerReportRenewInterval + + result, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}) + if err != nil { + t.Fatal(err) + } + first := producerEntries(t, c)["resolver-a"] + if first.RenewTime == nil || time.Since(first.RenewTime.Time) > time.Minute { + t.Fatalf("resolver-a renewTime = %v, want now", first.RenewTime) + } + if result.RequeueAfter != interval { + t.Fatalf("RequeueAfter after publishing = %v, want %v", result.RequeueAfter, interval) + } + + // An unchanged decision within the renew interval is not rewritten. + if err := c.Get(ctx, key, endpoint); err != nil { + t.Fatal(err) + } + version := endpoint.ResourceVersion + result, err = r.Reconcile(ctx, ctrl.Request{NamespacedName: key}) + if err != nil { + t.Fatal(err) + } + if err := c.Get(ctx, key, endpoint); err != nil { + t.Fatal(err) + } + if endpoint.ResourceVersion != version { + t.Fatal("an unchanged decision was rewritten within the renew interval") + } + if result.RequeueAfter <= 0 || result.RequeueAfter > interval { + t.Fatalf("RequeueAfter before renewal is due = %v, want within %v", result.RequeueAfter, interval) + } + + // Once the interval passes, the same decision is renewed. + later := first.RenewTime.Add(interval) + entries := []networkv1alpha1.ServiceEndpointProducerStatus{first} + entries[0].RenewTime = nil + if _, err := r.publish(ctx, endpoint, entries, later); err != nil { + t.Fatal(err) + } + if got := producerEntries(t, c)["resolver-a"]; got.RenewTime == nil || !got.RenewTime.Time.Equal(later) || + got.State != first.State || !got.LastTransitionTime.Equal(&first.LastTransitionTime) { + t.Fatalf("resolver-a after renewal = %+v, want the same decision renewed at %v", got, later) + } +} diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index ea82dfd8..c638be7c 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -15,6 +15,7 @@ import ( "sync" "time" + "k8s.io/apimachinery/pkg/api/equality" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -23,6 +24,7 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/builder" "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/manager" @@ -253,11 +255,26 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R logger.Error(err, "update ServiceRoutePolicy acceptance status") return ctrl.Result{}, err } + return serviceRouteRequeue(policy, endpoint, intents, time.Now()), nil +} + +// serviceRouteRequeue returns when a programmed policy is next checked: at the +// map resync interval while it has intents or a frontend, and no later than +// the next expiry of one of endpoint's producer reports, which can move +// consumers to another node's producer without any event, including onto a +// node with no intents yet. +func serviceRouteRequeue(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, + intents []serviceroute.RouteIntent, now time.Time, +) ctrl.Result { requeue := ctrl.Result{} if len(intents) != 0 || policy.Spec.Frontend != nil { requeue.RequeueAfter = serviceRouteMapResyncInterval } - return requeue, nil + if left, ok := serviceroute.NextProducerReportExpiry(endpoint, now); ok && + (requeue.RequeueAfter == 0 || left < requeue.RequeueAfter) { + requeue.RequeueAfter = left + } + return requeue } func (r *ServiceRoutePolicyReconciler) setAccepted( @@ -300,7 +317,12 @@ func (r *ServiceRoutePolicyReconciler) SetupWithManager(mgr ctrl.Manager) error For(&networkv1alpha1.ServiceRoutePolicy{}, builder.WithPredicates(predicate.GenerationChangedPredicate{})). Watches(&networkv1alpha1.ServiceRoutePolicy{}, handler.EnqueueRequestsFromMapFunc(r.attachmentPolicies), builder.WithPredicates(predicate.GenerationChangedPredicate{})). - Watches(&networkv1alpha1.ServiceEndpoint{}, handler.EnqueueRequestsFromMapFunc(r.endpointPolicies)). + // A renewal of producer reports that are still current changes no + // route; expiry is handled by requeueing. + Watches(&networkv1alpha1.ServiceEndpoint{}, handler.EnqueueRequestsFromMapFunc(r.endpointPolicies), + 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)). Watches(&networkv1alpha1.BGPRouter{}, handler.EnqueueRequestsFromMapFunc(r.routingDependencyPolicies)). @@ -513,6 +535,33 @@ func (r *ServiceRoutePolicyReconciler) consumerVPCIdentity( return vpc.Status.VPC, nil } +// producerReportsOnlyRenewed reports whether newObj differs from oldObj only +// in the renewTime of status.producers entries that were current at now. +func producerReportsOnlyRenewed(oldObj, newObj client.Object, now time.Time) bool { + oldEndpoint, ok := oldObj.(*networkv1alpha1.ServiceEndpoint) + if !ok { + return false + } + newEndpoint, ok := newObj.(*networkv1alpha1.ServiceEndpoint) + if !ok || len(oldEndpoint.Status.Producers) != len(newEndpoint.Status.Producers) { + return false + } + for _, entry := range oldEndpoint.Status.Producers { + if !serviceroute.ProducerReportCurrent(entry, now) { + return false + } + } + strip := func(endpoint *networkv1alpha1.ServiceEndpoint) *networkv1alpha1.ServiceEndpoint { + endpoint = endpoint.DeepCopy() + endpoint.ResourceVersion, endpoint.ManagedFields = "", nil + for i := range endpoint.Status.Producers { + endpoint.Status.Producers[i].RenewTime = nil + } + return endpoint + } + return equality.Semantic.DeepEqual(strip(oldEndpoint), strip(newEndpoint)) +} + func (r *ServiceRoutePolicyReconciler) endpointPolicies(ctx context.Context, obj client.Object) []ctrl.Request { return r.allPolicies(ctx) } diff --git a/internal/controller/serviceroute_spread_test.go b/internal/controller/serviceroute_spread_test.go index 99d7ba9b..b7c63338 100644 --- a/internal/controller/serviceroute_spread_test.go +++ b/internal/controller/serviceroute_spread_test.go @@ -7,9 +7,12 @@ package controller import ( "errors" "testing" + "time" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" "go.datum.net/galactic/internal/serviceroute" ) @@ -93,3 +96,45 @@ func TestServiceRouteReplacePolicyFallsBackWhenProducerSetsCannotCoexist(t *test } } } + +func TestProducerReportsOnlyRenewed(t *testing.T) { + now := time.Now() + renewedAt := func(ago time.Duration) *metav1.Time { + at := metav1.NewTime(now.Add(-ago)) + return &at + } + endpoint := func(state networkv1alpha1.BackendHealthState, renewed *metav1.Time) *networkv1alpha1.ServiceEndpoint { + return &networkv1alpha1.ServiceEndpoint{ + ObjectMeta: metav1.ObjectMeta{Name: producerHealthEndpoint, Namespace: producerHealthNamespace, + ResourceVersion: renewed.String()}, + Status: networkv1alpha1.ServiceEndpointStatus{Producers: []networkv1alpha1.ServiceEndpointProducerStatus{{ + Namespace: producerHealthNamespace, Name: "resolver-a", AttachmentUID: "uid-a", + Node: producerHealthNode, Port: 8053, + State: state, Reason: "ProbeSucceeded", RenewTime: renewed, + }}}, + } + } + lifetime := networkv1alpha1.ServiceEndpointProducerReportLifetime + for _, tc := range []struct { + name string + old, new *networkv1alpha1.ServiceEndpoint + want bool + }{ + {"current report renewed", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(20*time.Second)), + endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(0)), true}, + {"expired report renewed", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(lifetime+time.Second)), + endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(0)), false}, + {"decision changed", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(20*time.Second)), + endpoint(networkv1alpha1.BackendHealthUnhealthy, renewedAt(0)), false}, + } { + if got := producerReportsOnlyRenewed(tc.old, tc.new, now); got != tc.want { + t.Errorf("%s: producerReportsOnlyRenewed = %v, want %v", tc.name, got, tc.want) + } + } + added := endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(0)) + added.Status.Producers = append(added.Status.Producers, added.Status.Producers[0]) + added.Status.Producers[1].Name = "resolver-b" + if producerReportsOnlyRenewed(endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(time.Second)), added, now) { + t.Error("producerReportsOnlyRenewed accepted an added producer") + } +} diff --git a/internal/serviceroute/planner.go b/internal/serviceroute/planner.go index d79d4f6c..e5be5b65 100644 --- a/internal/serviceroute/planner.go +++ b/internal/serviceroute/planner.go @@ -146,7 +146,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph if err := validateProducerTopology(endpoint, producers); err != nil { return nil, err } - unhealthy := unhealthyProducers(endpoint, producers) + health := producersHealth(endpoint, producers, time.Now()) consumers := readyAttachments(attachments) intents := make([]RouteIntent, 0, len(consumers)) @@ -157,7 +157,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph if !consumerMatchesVPC(policy, consumer, consumerVPCIdentity) { continue } - local, producer := selectProducers(endpoint, consumer.Status.Node, producers, unhealthy) + local, producer := selectProducers(endpoint, consumer.Status.Node, producers, health) consumerKey := types.NamespacedName{Namespace: consumer.Namespace, Name: consumer.Name} if len(local) != 0 { if consumer.Status.Node == localNode { @@ -165,7 +165,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph Attachment: consumerKey, Kind: RouteIntentLocalSet, Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ConsumerDevice: consumer.Status.HostInterface, Ports: clonePorts(ports), ConsumerNextHop: attachmentNextHop(consumer, service.IP), - Producers: producerTargets(local, unhealthy, service.IP), + Producers: producerTargets(local, health, service.IP), }) } continue @@ -344,10 +344,11 @@ func spread(endpoint *networkv1alpha1.ServiceEndpoint) bool { // selectProducers returns, for a Spread endpoint, every producer on // consumerNode as local. Otherwise, or when consumerNode has none, it returns // the one producer a consumer there is sent to, or nil. A Spread endpoint's -// remote producer is the first one not Unhealthy, or the first one when all -// are. +// remote producer is the first one whose node currently reports it Healthy or +// Unknown, or has never reported it; failing that, the first one whose report +// expired, since its node may be down; failing that, the first one. func selectProducers(endpoint *networkv1alpha1.ServiceEndpoint, consumerNode string, - producers []*cloudv1alpha1.VPCAttachment, unhealthy map[types.UID]bool, + producers []*cloudv1alpha1.VPCAttachment, health map[types.UID]producerHealth, ) (local []*cloudv1alpha1.VPCAttachment, remote *cloudv1alpha1.VPCAttachment) { if spread(endpoint) { for _, producer := range producers { @@ -362,9 +363,11 @@ func selectProducers(endpoint *networkv1alpha1.ServiceEndpoint, consumerNode str len(producers) == 0 { return nil, nil } - for _, producer := range producers { - if !unhealthy[producer.UID] { - return nil, producer + for _, want := range []producerHealth{producerEligible, producerReportExpired} { + for _, producer := range producers { + if health[producer.UID] == want { + return nil, producer + } } } return nil, producers[0] @@ -386,12 +389,26 @@ func chooseProducer(mode networkv1alpha1.ServiceEndpointDeliveryMode, consumerNo return nil } -// unhealthyProducers returns the UIDs of a Spread endpoint's producers whose -// status entry, written by their own node for that same attachment, says -// Unhealthy. A producer with no current entry is not in it. -func unhealthyProducers(endpoint *networkv1alpha1.ServiceEndpoint, - producers []*cloudv1alpha1.VPCAttachment, -) map[types.UID]bool { +// producerHealth is what a Spread endpoint's status.producers says of one +// producer. +type producerHealth int + +const ( + // producerEligible: its node currently reports it Healthy or Unknown, or + // has no entry for it. + producerEligible producerHealth = iota + // producerReportExpired: its node stopped renewing its entry. + producerReportExpired + // producerUnhealthy: its node currently reports it Unhealthy. + producerUnhealthy +) + +// producersHealth classifies a Spread endpoint's producers by the status entry +// their own node wrote for that same attachment, as of now. A producer with +// no such entry is eligible. +func producersHealth(endpoint *networkv1alpha1.ServiceEndpoint, + producers []*cloudv1alpha1.VPCAttachment, now time.Time, +) map[types.UID]producerHealth { if !spread(endpoint) { return nil } @@ -400,26 +417,54 @@ func unhealthyProducers(endpoint *networkv1alpha1.ServiceEndpoint, for _, entry := range endpoint.Status.Producers { reported[types.NamespacedName{Namespace: entry.Namespace, Name: entry.Name}] = entry } - unhealthy := make(map[types.UID]bool) + health := make(map[types.UID]producerHealth, len(producers)) for _, producer := range producers { entry, ok := reported[types.NamespacedName{Namespace: producer.Namespace, Name: producer.Name}] - if ok && entry.AttachmentUID == producer.UID && entry.Node == producer.Status.Node && - entry.State == networkv1alpha1.BackendHealthUnhealthy { - unhealthy[producer.UID] = true + switch { + case !ok || entry.AttachmentUID != producer.UID || entry.Node != producer.Status.Node: + case !ProducerReportCurrent(entry, now): + health[producer.UID] = producerReportExpired + case entry.State == networkv1alpha1.BackendHealthUnhealthy: + health[producer.UID] = producerUnhealthy + } + } + return health +} + +// ProducerReportCurrent reports whether entry was renewed within +// ServiceEndpointProducerReportLifetime before now. +func ProducerReportCurrent(entry networkv1alpha1.ServiceEndpointProducerStatus, now time.Time) bool { + return entry.RenewTime != nil && + now.Before(entry.RenewTime.Add(networkv1alpha1.ServiceEndpointProducerReportLifetime)) +} + +// NextProducerReportExpiry returns how long until the first current entry of +// endpoint's status.producers expires, and false when none is current. +func NextProducerReportExpiry(endpoint *networkv1alpha1.ServiceEndpoint, now time.Time) (time.Duration, bool) { + var next time.Duration + found := false + for _, entry := range endpoint.Status.Producers { + if !ProducerReportCurrent(entry, now) { + continue + } + left := entry.RenewTime.Add(networkv1alpha1.ServiceEndpointProducerReportLifetime).Sub(now) + if !found || left < next { + next, found = left, true } } - return unhealthy + return next, found } -// producerTargets builds a LocalSet intent's members. Unhealthy producers -// drain, unless every one of them is Unhealthy: then all stay eligible, since -// an empty set would drop every new flow. -func producerTargets(local []*cloudv1alpha1.VPCAttachment, unhealthy map[types.UID]bool, +// producerTargets builds a LocalSet intent's members. Producers currently +// reported Unhealthy drain, unless every one of them is: then all stay +// eligible, since an empty set would drop every new flow. An expired report +// drains nothing. +func producerTargets(local []*cloudv1alpha1.VPCAttachment, health map[types.UID]producerHealth, service net.IP, ) []ProducerTarget { allUnhealthy := true for _, producer := range local { - if !unhealthy[producer.UID] { + if health[producer.UID] != producerUnhealthy { allUnhealthy = false break } @@ -430,7 +475,7 @@ func producerTargets(local []*cloudv1alpha1.VPCAttachment, unhealthy map[types.U Attachment: types.NamespacedName{Namespace: producer.Namespace, Name: producer.Name}, Device: producer.Status.HostInterface, NextHop: attachmentNextHop(producer, service), - Draining: unhealthy[producer.UID] && !allUnhealthy, + Draining: health[producer.UID] == producerUnhealthy && !allUnhealthy, }) } return targets diff --git a/internal/serviceroute/planner_spread_test.go b/internal/serviceroute/planner_spread_test.go index 9ca2ae39..65c06531 100644 --- a/internal/serviceroute/planner_spread_test.go +++ b/internal/serviceroute/planner_spread_test.go @@ -6,11 +6,14 @@ package serviceroute import ( "fmt" "net" + "slices" "strings" "testing" + "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/types" + "k8s.io/utils/ptr" cloudv1alpha1 "go.datum.net/galactic/api/cloud/v1alpha1" networkv1alpha1 "go.datum.net/galactic/api/network/v1alpha1" @@ -20,6 +23,7 @@ const ( spreadProducerNamespace = "service" spreadConsumerName = "consumer" bothLocalProducers = "producer-a,producer-z" + firstRemoteProducer = "a-remote" ) func spreadEndpoint(mode networkv1alpha1.ServiceEndpointDeliveryMode) ( @@ -42,9 +46,18 @@ func reportHealth(endpoint *networkv1alpha1.ServiceEndpoint, producer *cloudv1al endpoint.Status.Producers = append(endpoint.Status.Producers, networkv1alpha1.ServiceEndpointProducerStatus{ Namespace: producer.Namespace, Name: producer.Name, AttachmentUID: producer.UID, Node: producer.Status.Node, Port: 8053, State: state, LastTransitionTime: metav1.Now(), + RenewTime: ptr.To(metav1.Now()), }) } +// expireReport makes the last entry of endpoint's status.producers one its +// node stopped renewing before the report lifetime. +func expireReport(endpoint *networkv1alpha1.ServiceEndpoint) { + entry := &endpoint.Status.Producers[len(endpoint.Status.Producers)-1] + entry.RenewTime = ptr.To(metav1.NewTime( + time.Now().Add(-networkv1alpha1.ServiceEndpointProducerReportLifetime - time.Second))) +} + func producerSummary(intent RouteIntent) string { parts := make([]string, 0, len(intent.Producers)) for _, p := range intent.Producers { @@ -136,7 +149,7 @@ func TestCompileSpreadDrainsUnhealthyProducers(t *testing.T) { func TestCompileSpreadFallsBackToHealthyRemoteProducer(t *testing.T) { policy, endpoint := spreadEndpoint(networkv1alpha1.ServiceEndpointDeliveryModePreferNodeLocal) consumer := spreadConsumer() - first := readyAttachment(spreadProducerNamespace, "a-remote", "first-uid", testRemoteNodeName, "remote-a", + first := readyAttachment(spreadProducerNamespace, firstRemoteProducer, "first-uid", testRemoteNodeName, "remote-a", testLabels(testProducerLabel, "yes")) last := readyAttachment(spreadProducerNamespace, "z-remote", "last-uid", testRemoteNodeName, "remote-z", testLabels(testProducerLabel, "yes")) @@ -156,7 +169,7 @@ func TestCompileSpreadFallsBackToHealthyRemoteProducer(t *testing.T) { return got[0].ProducerAttachment } - if got := remote(); got.Name != "a-remote" { + if got := remote(); got.Name != firstRemoteProducer { t.Fatalf("remote producer = %s, want a-remote", got) } reportHealth(endpoint, first, networkv1alpha1.BackendHealthUnhealthy) @@ -164,7 +177,7 @@ func TestCompileSpreadFallsBackToHealthyRemoteProducer(t *testing.T) { t.Fatalf("remote producer with a-remote Unhealthy = %s, want z-remote", got) } reportHealth(endpoint, last, networkv1alpha1.BackendHealthUnhealthy) - if got := remote(); got.Name != "a-remote" { + if got := remote(); got.Name != firstRemoteProducer { t.Fatalf("remote producer with both Unhealthy = %s, want a-remote", got) } @@ -178,6 +191,120 @@ func TestCompileSpreadFallsBackToHealthyRemoteProducer(t *testing.T) { } } +func TestCompileSpreadIgnoresExpiredLocalReports(t *testing.T) { + policy, endpoint := spreadEndpoint(networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal) + consumer := spreadConsumer() + first := readyAttachment(spreadProducerNamespace, "a-producer", "first-uid", testNodeName, "producer-a", + testLabels(testProducerLabel, "yes")) + last := readyAttachment(spreadProducerNamespace, "z-producer", "last-uid", testNodeName, "producer-z", + testLabels(testProducerLabel, "yes")) + attachments := []*cloudv1alpha1.VPCAttachment{consumer, first, last} + compile := func() string { + t.Helper() + got, err := Compile(policy, endpoint, attachments, testNodeName, nil) + if err != nil { + t.Fatalf("Compile: %v", err) + } + if len(got) != 1 { + t.Fatalf("intents = %+v, want one", got) + } + return producerSummary(got[0]) + } + + // An Unhealthy decision written before its node's router stopped no + // longer drains the producer. + reportHealth(endpoint, first, networkv1alpha1.BackendHealthUnhealthy) + expireReport(endpoint) + reportHealth(endpoint, last, networkv1alpha1.BackendHealthHealthy) + if got := compile(); got != bothLocalProducers { + t.Fatalf("producers with an expired Unhealthy decision = %s, want all eligible", got) + } + endpoint.Status.Producers[0].RenewTime = nil + if got := compile(); got != bothLocalProducers { + t.Fatalf("producers with a never-renewed Unhealthy decision = %s, want all eligible", got) + } +} + +func TestCompileSpreadAvoidsRemoteProducersWithExpiredReports(t *testing.T) { + policy, endpoint := spreadEndpoint(networkv1alpha1.ServiceEndpointDeliveryModePreferNodeLocal) + consumer := spreadConsumer() + lost := readyAttachment(spreadProducerNamespace, firstRemoteProducer, "lost-uid", testRemoteNodeName, "remote-a", + testLabels(testProducerLabel, "yes")) + live := readyAttachment(spreadProducerNamespace, "z-remote", "live-uid", "node-c", "remote-z", + testLabels(testProducerLabel, "yes")) + resolver := func(attachment *cloudv1alpha1.VPCAttachment) (net.IP, error) { + return net.ParseIP("fd00::" + string(attachment.UID[0])), nil + } + attachments := []*cloudv1alpha1.VPCAttachment{consumer, lost, live} + remote := func() string { + t.Helper() + got, err := Compile(policy, endpoint, attachments, testNodeName, resolver) + if err != nil { + t.Fatalf("Compile: %v", err) + } + if len(got) != 1 || got[0].Kind != RouteIntentRemoteConsumer { + t.Fatalf("intents = %+v, want one RemoteConsumer intent", got) + } + plan, err := ProgrammingPlan(policy, endpoint, attachments, "") + if err != nil { + t.Fatalf("ProgrammingPlan: %v", err) + } + if want := []string{got[0].ProducerAttachment.Name}; !planNodesInclude(plan, attachments, want) { + t.Fatalf("ProgrammingPlan nodes = %v, want the node of %v", plan.Nodes, want) + } + return got[0].ProducerAttachment.Name + } + + // a-remote's node went down while its attachment still reads ready. + reportHealth(endpoint, lost, networkv1alpha1.BackendHealthHealthy) + expireReport(endpoint) + reportHealth(endpoint, live, networkv1alpha1.BackendHealthHealthy) + if got := remote(); got != "z-remote" { + t.Fatalf("remote producer with a-remote's report expired = %s, want z-remote", got) + } + // An expired report is preferred over a current Unhealthy one. + endpoint.Status.Producers[1].State = networkv1alpha1.BackendHealthUnhealthy + if got := remote(); got != firstRemoteProducer { + t.Fatalf("remote producer with z-remote Unhealthy = %s, want a-remote", got) + } + endpoint.Status.Producers[1].State = networkv1alpha1.BackendHealthHealthy + expireReport(endpoint) + if got := remote(); got != firstRemoteProducer { + t.Fatalf("remote producer with every report expired = %s, want a-remote", got) + } +} + +func TestNextProducerReportExpiry(t *testing.T) { + _, endpoint := spreadEndpoint(networkv1alpha1.ServiceEndpointDeliveryModePreferNodeLocal) + now := time.Now() + if _, ok := NextProducerReportExpiry(endpoint, now); ok { + t.Fatal("NextProducerReportExpiry found an expiry with no reports") + } + lifetime := networkv1alpha1.ServiceEndpointProducerReportLifetime + for _, ago := range []time.Duration{lifetime + time.Second, 10 * time.Second, 40 * time.Second} { + renewed := metav1.NewTime(now.Add(-ago)) + endpoint.Status.Producers = append(endpoint.Status.Producers, + networkv1alpha1.ServiceEndpointProducerStatus{RenewTime: &renewed}) + } + endpoint.Status.Producers = append(endpoint.Status.Producers, networkv1alpha1.ServiceEndpointProducerStatus{}) + if left, ok := NextProducerReportExpiry(endpoint, now); !ok || left != lifetime-40*time.Second { + t.Fatalf("NextProducerReportExpiry = %v, %v, want %v", left, ok, lifetime-40*time.Second) + } +} + +func planNodesInclude(plan ProgrammingRequirements, attachments []*cloudv1alpha1.VPCAttachment, + names []string, +) bool { + for _, attachment := range attachments { + for _, name := range names { + if attachment.Name == name && !slices.Contains(plan.Nodes, attachment.Status.Node) { + return false + } + } + } + return true +} + func TestCompileSpreadRefusesMoreProducersThanASetHolds(t *testing.T) { policy, endpoint := spreadEndpoint(networkv1alpha1.ServiceEndpointDeliveryModeNodeLocal) attachments := make([]*cloudv1alpha1.VPCAttachment, 0, networkv1alpha1.ServiceEndpointMaxProducersPerNode+2) diff --git a/internal/serviceroute/programming.go b/internal/serviceroute/programming.go index 24e5d93b..add320c3 100644 --- a/internal/serviceroute/programming.go +++ b/internal/serviceroute/programming.go @@ -10,6 +10,7 @@ import ( "errors" "fmt" "sort" + "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/labels" @@ -48,7 +49,7 @@ func ProgrammingPlan(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *netwo if err := validateProducerTopology(endpoint, producers); err != nil { return plan, err } - unhealthy := unhealthyProducers(endpoint, producers) + health := producersHealth(endpoint, producers, time.Now()) ready := map[types.NamespacedName]bool{} for _, a := range readyAttachments(attachments) { ready[types.NamespacedName{Namespace: a.Namespace, Name: a.Name}] = a.Status.HostInterface != "" @@ -80,7 +81,7 @@ func ProgrammingPlan(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *netwo continue } selected++ - local, p := selectProducers(endpoint, a.Status.Node, producers, unhealthy) + local, p := selectProducers(endpoint, a.Status.Node, producers, health) if len(local) != 0 { p = local[0] } From 7b2c8f87fae9ab0b203813870618389707125f7f Mon Sep 17 00:00:00 2001 From: Peter Sprygada Date: Tue, 27 Oct 2026 12:45:45 -0400 Subject: [PATCH 2/2] fix: Reconcile when a producer report loses its renewal The ServiceEndpoint watch filter skipped any update that changed only the renewTime of current status.producers entries. An update that clears renewTime, as a rolled-back router's apply does, or that leaves it expired, turns a current entry into an expired one, and the filter treated it as a renewal, so the change waited for the next expiry requeue. The filter now passes an update unless every entry is current both before and after it. The ServiceEndpoint CRD is regenerated with the repository's controller-gen v0.21.0, so its version annotation matches the other CRDs. The private services guide states that the CRD has to be applied before galactic-router is upgraded, since the API server otherwise prunes renewTime from every report, and that node clocks have to stay in sync, since each node compares renewTime with its own clock. Related to #1099 --- .../crd/galactic.datumapis.com_serviceendpoints.yaml | 2 +- docs/private-services/README.md | 7 +++++++ internal/controller/serviceroute_controller.go | 11 +++++++---- internal/controller/serviceroute_spread_test.go | 4 +++- 4 files changed, 18 insertions(+), 6 deletions(-) diff --git a/config/crd/galactic.datumapis.com_serviceendpoints.yaml b/config/crd/galactic.datumapis.com_serviceendpoints.yaml index 380d7c90..4fda70eb 100644 --- a/config/crd/galactic.datumapis.com_serviceendpoints.yaml +++ b/config/crd/galactic.datumapis.com_serviceendpoints.yaml @@ -3,7 +3,7 @@ apiVersion: apiextensions.k8s.io/v1 kind: CustomResourceDefinition metadata: annotations: - controller-gen.kubebuilder.io/version: v0.18.0 + controller-gen.kubebuilder.io/version: v0.21.0 name: serviceendpoints.galactic.datumapis.com spec: group: galactic.datumapis.com diff --git a/docs/private-services/README.md b/docs/private-services/README.md index 8581f67f..35608743 100644 --- a/docs/private-services/README.md +++ b/docs/private-services/README.md @@ -148,6 +148,13 @@ nodes that are not yet upgraded as expired, and send remote consumers to them only when no current entry offers a producer. This lasts until the rollout finishes. A rolled-back router ignores `renewTime`. +Apply the `ServiceEndpoint` CRD with `renewTime` before upgrading +`galactic-router`. Without it the API server prunes the field from every +report, so every entry reads as expired and no `Unhealthy` producer drains. +Nodes compare `renewTime` with their own clock, so keep node clocks in sync: +a skew near 45 seconds expires live reports early or keeps a lost node's +reports current for that much longer. + Spread needs a `galactic-cni` whose datapath has the producer set map. A `galactic-router` that finds the map missing refuses the endpoint's Spread routes on that node, retrying, and keeps every other route working. Roll out diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index c638be7c..ecd0cfda 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -536,7 +536,8 @@ func (r *ServiceRoutePolicyReconciler) consumerVPCIdentity( } // producerReportsOnlyRenewed reports whether newObj differs from oldObj only -// in the renewTime of status.producers entries that were current at now. +// in the renewTime of status.producers entries that are current at now in +// both. func producerReportsOnlyRenewed(oldObj, newObj client.Object, now time.Time) bool { oldEndpoint, ok := oldObj.(*networkv1alpha1.ServiceEndpoint) if !ok { @@ -546,9 +547,11 @@ func producerReportsOnlyRenewed(oldObj, newObj client.Object, now time.Time) boo if !ok || len(oldEndpoint.Status.Producers) != len(newEndpoint.Status.Producers) { return false } - for _, entry := range oldEndpoint.Status.Producers { - if !serviceroute.ProducerReportCurrent(entry, now) { - return false + for _, endpoint := range []*networkv1alpha1.ServiceEndpoint{oldEndpoint, newEndpoint} { + for _, entry := range endpoint.Status.Producers { + if !serviceroute.ProducerReportCurrent(entry, now) { + return false + } } } strip := func(endpoint *networkv1alpha1.ServiceEndpoint) *networkv1alpha1.ServiceEndpoint { diff --git a/internal/controller/serviceroute_spread_test.go b/internal/controller/serviceroute_spread_test.go index b7c63338..c1282167 100644 --- a/internal/controller/serviceroute_spread_test.go +++ b/internal/controller/serviceroute_spread_test.go @@ -106,7 +106,7 @@ func TestProducerReportsOnlyRenewed(t *testing.T) { endpoint := func(state networkv1alpha1.BackendHealthState, renewed *metav1.Time) *networkv1alpha1.ServiceEndpoint { return &networkv1alpha1.ServiceEndpoint{ ObjectMeta: metav1.ObjectMeta{Name: producerHealthEndpoint, Namespace: producerHealthNamespace, - ResourceVersion: renewed.String()}, + ResourceVersion: "1"}, Status: networkv1alpha1.ServiceEndpointStatus{Producers: []networkv1alpha1.ServiceEndpointProducerStatus{{ Namespace: producerHealthNamespace, Name: "resolver-a", AttachmentUID: "uid-a", Node: producerHealthNode, Port: 8053, @@ -124,6 +124,8 @@ func TestProducerReportsOnlyRenewed(t *testing.T) { endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(0)), true}, {"expired report renewed", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(lifetime+time.Second)), endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(0)), false}, + {"renewTime dropped", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(20*time.Second)), + endpoint(networkv1alpha1.BackendHealthHealthy, nil), false}, {"decision changed", endpoint(networkv1alpha1.BackendHealthHealthy, renewedAt(20*time.Second)), endpoint(networkv1alpha1.BackendHealthUnhealthy, renewedAt(0)), false}, } {