diff --git a/api/network/v1alpha1/serviceroute_admission_envtest_test.go b/api/network/v1alpha1/serviceroute_admission_envtest_test.go index c149ef2a..c7303db1 100644 --- a/api/network/v1alpha1/serviceroute_admission_envtest_test.go +++ b/api/network/v1alpha1/serviceroute_admission_envtest_test.go @@ -94,6 +94,15 @@ func TestTranslatedAPIAdmission(t *testing.T) { p.Spec.Authorization = nil }, false}, {"translated", func(*networkv1alpha1.ServiceRoutePolicy) {}, false}, + {"persistent", func(p *networkv1alpha1.ServiceRoutePolicy) { + p.Spec.Authorization = &networkv1alpha1.ServiceRouteAuthorization{Persistent: true} + }, false}, + {"persistent-with-deadline", func(p *networkv1alpha1.ServiceRoutePolicy) { + p.Spec.Authorization.Persistent = true + }, true}, + {"missing-deadline", func(p *networkv1alpha1.ServiceRoutePolicy) { + p.Spec.Authorization = &networkv1alpha1.ServiceRouteAuthorization{} + }, true}, } { t.Run(tt.name, func(t *testing.T) { p := policy(tt.name) diff --git a/api/network/v1alpha1/serviceroute_types.go b/api/network/v1alpha1/serviceroute_types.go index 07525a4a..518e1ef7 100644 --- a/api/network/v1alpha1/serviceroute_types.go +++ b/api/network/v1alpha1/serviceroute_types.go @@ -146,8 +146,7 @@ type ServiceRoutePolicySpec struct { // +optional Frontend *ServiceRouteFrontend `json:"frontend,omitempty"` - // Authorization bounds translated access independently of application leases. - // Only trusted networking integration may renew this deadline. + // Authorization defines the lifetime of translated service access. // +optional Authorization *ServiceRouteAuthorization `json:"authorization,omitempty"` @@ -169,11 +168,17 @@ type ServiceRoutePolicySpec struct { Region string `json:"region,omitempty"` } -// ServiceRouteAuthorization bounds one network authorization episode. +// ServiceRouteAuthorization defines persistent or time-bounded access. +// +kubebuilder:validation:XValidation:rule="has(self.persistent) && self.persistent ? !has(self.validUntil) : has(self.validUntil)",message="persistent access must omit validUntil; bounded access requires validUntil" type ServiceRouteAuthorization struct { + // Persistent retains access until policy withdrawal or VPC lifetime replacement. + // +optional + Persistent bool `json:"persistent,omitempty"` + // ValidUntil is the absolute UTC deadline for requests and existing replies. // Node controllers reject deadlines more than two minutes in the future. - ValidUntil metav1.Time `json:"validUntil"` + // +optional + ValidUntil metav1.Time `json:"validUntil,omitempty"` } // ServiceRouteVPCReference identifies one immutable Cloud API VPC lifetime. diff --git a/config/crd/galactic.datumapis.com_serviceroutepolicies.yaml b/config/crd/galactic.datumapis.com_serviceroutepolicies.yaml index 759b18ee..9906dc3e 100644 --- a/config/crd/galactic.datumapis.com_serviceroutepolicies.yaml +++ b/config/crd/galactic.datumapis.com_serviceroutepolicies.yaml @@ -105,19 +105,25 @@ spec: type: object x-kubernetes-map-type: atomic authorization: - description: |- - Authorization bounds translated access independently of application leases. - Only trusted networking integration may renew this deadline. + description: Authorization defines the lifetime of translated service + access. properties: + persistent: + description: Persistent retains access until policy withdrawal + or VPC lifetime replacement. + type: boolean validUntil: description: |- ValidUntil is the absolute UTC deadline for requests and existing replies. Node controllers reject deadlines more than two minutes in the future. format: date-time type: string - required: - - validUntil type: object + x-kubernetes-validations: + - message: persistent access must omit validUntil; bounded access + requires validUntil + rule: 'has(self.persistent) && self.persistent ? !has(self.validUntil) + : has(self.validUntil)' consumerVPCRef: description: |- ConsumerVPCRef pins authorization to the immutable lifetime of the VPC diff --git a/config/monitoring/prometheusrule.yaml b/config/monitoring/prometheusrule.yaml index 6df7d20b..cf94dc13 100644 --- a/config/monitoring/prometheusrule.yaml +++ b/config/monitoring/prometheusrule.yaml @@ -496,6 +496,20 @@ spec: {{ $labels.kind }} routes it holds a best path for, so this node cannot reach those prefixes. The router's "route install failed" log lines name each prefix and the error, usually a SID with no route over an SRv6 uplink or an unresolved neighbor. + - alert: GalacticServiceRouteProgrammingFailing + expr: sum without (operation, result) (increase(galactic_service_route_route_operations_total{job="galactic-router",result="error"}[10m])) > 0 + for: 15m + labels: + severity: warning + service: galactic + team: connect + annotations: + summary: "Private service routing on {{ $labels.node }} keeps failing" + runbook_url: "https://github.com/datum-cloud/infra/blob/main/docs/runbooks/galactic/README.md#galacticservicerouteprogrammingfailing" + description: >- + Private service routes have repeatedly failed to apply or remove on {{ $labels.node }}. + Consumers can lose access even when the router Pod is ready and policies are accepted. + Check the router's service policy errors, attachment host interfaces, and node programming acknowledgments. - alert: GalacticRouterMetricsDown expr: up{job=~"galactic-router|galactic-router-rr"} == 0 for: 5m diff --git a/config/monitoring/tests/rules-test.yaml b/config/monitoring/tests/rules-test.yaml index 8d1f43c5..2f6d4241 100644 --- a/config/monitoring/tests/rules-test.yaml +++ b/config/monitoring/tests/rules-test.yaml @@ -2289,3 +2289,33 @@ tests: galactic-nat on dfw-worker2 has failed to renew its nat slot's lease in the XDP dispatcher more than twice in every 5-minute window for 5 minutes. Once the lease expires the dispatcher skips the slot and tenant traffic leaves untranslated. The shard's log names the error. + + # Repeated failures alert; a rollout burst and idle or disabled routing do not. + - interval: 30s + input_series: + - series: 'galactic_service_route_route_operations_total{job="galactic-router",node="failing",operation="apply",result="error"}' + values: "0+1x60" + - series: 'galactic_service_route_route_operations_total{job="galactic-router",node="recovering",operation="apply",result="error"}' + values: "0+1x5 5x55" + - series: 'galactic_service_route_route_operations_total{job="galactic-router",node="idle",operation="apply",result="error"}' + values: "0x60" + alert_rule_test: + - eval_time: 10m + alertname: GalacticServiceRouteProgrammingFailing + exp_alerts: [] + - eval_time: 25m + alertname: GalacticServiceRouteProgrammingFailing + exp_alerts: + - exp_labels: + severity: warning + service: galactic + team: connect + job: galactic-router + node: failing + exp_annotations: + summary: "Private service routing on failing keeps failing" + runbook_url: "https://github.com/datum-cloud/infra/blob/main/docs/runbooks/galactic/README.md#galacticservicerouteprogrammingfailing" + description: >- + Private service routes have repeatedly failed to apply or remove on failing. + Consumers can lose access even when the router Pod is ready and policies are accepted. + Check the router's service policy errors, attachment host interfaces, and node programming acknowledgments. diff --git a/docs/api/network-gateway.md b/docs/api/network-gateway.md index 94caab46..c744941e 100644 --- a/docs/api/network-gateway.md +++ b/docs/api/network-gateway.md @@ -574,7 +574,7 @@ _Appears in:_ -ServiceRouteAuthorization bounds one network authorization episode. +ServiceRouteAuthorization defines persistent or time-bounded access. @@ -583,6 +583,7 @@ _Appears in:_ | Field | Description | Default | Validation | | --- | --- | --- | --- | +| `persistent` _boolean_ | Persistent retains access until policy withdrawal or VPC lifetime replacement. | | | | `validUntil` _[Time](https://kubernetes.io/docs/reference/generated/kubernetes-api/v/#time-v1-meta)_ | ValidUntil is the absolute UTC deadline for requests and existing replies.
Node controllers reject deadlines more than two minutes in the future. | | | @@ -662,7 +663,7 @@ _Appears in:_ | `serviceRef` _[ServiceEndpointReference](#serviceendpointreference)_ | ServiceRef identifies the ServiceEndpoint in this namespace. | | Required: \{\}
| | `consumerVPCRef` _[ServiceRouteVPCReference](#serviceroutevpcreference)_ | ConsumerVPCRef pins authorization to the immutable lifetime of the VPC
selected by AttachmentSelector. Galactic resolves the name through the
Cloud API and requires both the live name and UID to match. | | | | `frontend` _[ServiceRouteFrontend](#serviceroutefrontend)_ | Frontend is the address presented inside the consumer VPC. When set,
Galactic translates it to the ServiceEndpoint address and restores it on
replies. | | | -| `authorization` _[ServiceRouteAuthorization](#servicerouteauthorization)_ | Authorization bounds translated access independently of application leases.
Only trusted networking integration may renew this deadline. | | | +| `authorization` _[ServiceRouteAuthorization](#servicerouteauthorization)_ | Authorization defines the lifetime of translated service access. | | | | `attachmentSelector` _[LabelSelector](https://kubernetes.io/docs/reference/generated/kubernetes-api/v/#labelselector-v1-meta)_ | AttachmentSelector selects eligible VPC attachments by authoritative
labels. Labels that grant access to platform services must be assigned by
the network control plane, not by tenants. | | Required: \{\}
| | `protocolPorts` _[ServiceRouteProtocolPort](#servicerouteprotocolport) array_ | ProtocolPorts limits the traffic that the route is intended to carry.
An empty list means the endpoint's declared port and protocol are used. | | | | `region` _string_ | Region limits selected attachments to a region. An empty value means the
policy applies in every region where the endpoint is available. | | | diff --git a/docs/enhancements/networking/private-service-connect/README.md b/docs/enhancements/networking/private-service-connect/README.md index 8183980a..2622b1f6 100644 --- a/docs/enhancements/networking/private-service-connect/README.md +++ b/docs/enhancements/networking/private-service-connect/README.md @@ -389,7 +389,8 @@ Service owners withdraw unhealthy producers. Networking integration deletes access when the consumer or service is deleted. Galactic removes obsolete programming and reconciles current lifetimes after restart. Recreated VPCs and endpoints must not inherit previous grants. The current route policy has no -authorization lease; rollout requires a bounded revocation policy. +authorization policy. Persistent access requires explicit revocation; bounded +access requires a renewed deadline. ### Data plane @@ -443,7 +444,7 @@ networking integration maps that frontend to an authorized service-side destination for the network's DNS context. Galactic enforces the private path; DNS interprets the destination and isolates zones and answers. -DNS contexts, record publication, health-aware discovery, and resolver leases +DNS contexts, record publication, health-aware discovery, and resolver access belong to the [DNS component design](https://github.com/datum-cloud/dns-operator/blob/docs/internal-dns-architecture/docs/architecture/internal-dns/README.md). The generic private-service path does not interpret DNS resources. diff --git a/docs/router/private-service-routes.md b/docs/router/private-service-routes.md index bfce4814..21480c1a 100644 --- a/docs/router/private-service-routes.md +++ b/docs/router/private-service-routes.md @@ -75,16 +75,25 @@ The publishing controller supplies the producer address in `ServiceEndpoint`; `ServiceRoutePolicy.spec.frontend.address` supplies the consumer address. Both addresses must use the same IP family. Ports are not translated. -A translated policy must pin the consumer VPC by name and UID and include an -`authorization.validUntil` deadline within the next two minutes. Selected -attachments must match that VPC's allocated network identity. Kernel checks -stop new requests and established replies when the authorization expires, even -if the controller or its API is unavailable. Renew authorization explicitly to -restore access. +A translated policy pins the consumer VPC by name and UID. Selected attachments +must match that VPC's allocated network identity. Persistent access remains +active during control plane outages and ends when the policy is withdrawn or +the VPC lifetime changes. + +```yaml +spec: + authorization: + # Keep connectivity until an explicit policy or VPC lifetime change. + persistent: true +``` + +Time-bounded access sets `authorization.validUntil` within the next two minutes +and omits `persistent`. Kernel checks enforce its deadline on requests and +established replies. Persistent access must omit the deadline. Policies claiming the same frontend address, protocol, and port on one consumer attachment are rejected together. Separate VPC attachments can reuse that tuple. -The publisher owns authorization renewal and service health; Galactic enforces +The publisher owns access policy and service health; Galactic enforces network access and forwards packets. It does not interpret DNS records or zones. ## Node programming reports @@ -95,8 +104,9 @@ and input digest. It expires within 60 seconds or when authorization expires, whichever comes first. Consumer and producer nodes report independently for remote delivery. -Controllers using these reports must require the current topology digest and -all required nodes. `Accepted=True` means the policy is valid; it does not prove +Controllers require the current topology digest and all required nodes to +establish a new path. Expired reports indicate unknown observation health; they +do not revoke a persistent path. `Accepted=True` means the policy is valid; it does not prove that forwarding is installed. Reports cover network programming, not the health of the producer application. Startup withdraws previous local readiness before rebuilding the node's state. diff --git a/go.mod b/go.mod index eac1c70f..3345eb97 100644 --- a/go.mod +++ b/go.mod @@ -89,7 +89,6 @@ require ( github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect - github.com/onsi/ginkgo/v2 v2.28.2 // indirect github.com/orcaman/concurrent-map/v2 v2.0.1 // indirect github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/pkg/errors v0.9.1 // indirect diff --git a/go.sum b/go.sum index d8fe931b..ea2df73c 100644 --- a/go.sum +++ b/go.sum @@ -145,6 +145,7 @@ github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFd github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= github.com/onsi/ginkgo/v2 v2.28.2 h1:DTrMfpqxiNUyQ3Y0zhn1n3cOO2euFgQPYIpkWwxVFps= github.com/onsi/ginkgo/v2 v2.28.2/go.mod h1:CLtbVInNckU3/+gC8LzkGUb9oF+e8W8TdUsxPwvdOgE= github.com/onsi/gomega v1.42.1 h1:iN1rCUX+44NZ1Dc97MPoeFYbFR0vh8zxoxMFwKdyZ6I= @@ -198,24 +199,16 @@ github.com/stretchr/testify v1.12.1 h1:EuwCh5fleGS7H32xRwO3wRGT7DxrDhLAT6FF8MpWD github.com/stretchr/testify v1.12.1/go.mod h1:MDEgiDPPsNp5cuIrHPPCyornHKgEVbtFUmoNlxoYthg= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= -github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c h1:88GLXNKgc6pwIU/4yq+gleF5ISv3AsJkYmFlS1GjHvo= -github.com/vishvananda/netlink v1.3.2-0.20261002201547-1d1c62507e3c/go.mod h1:lEui7SPMd9fgxzHVGRAvTxsBGCF6PRH81o2kLWLWHgw= github.com/vishvananda/netlink v1.3.2-0.20261015184244-3fdd74264153 h1:h+5tbko9K2pYcDASxNj0GYkEJnWmdtpF0GN7PSuY9oM= github.com/vishvananda/netlink v1.3.2-0.20261015184244-3fdd74264153/go.mod h1:lEui7SPMd9fgxzHVGRAvTxsBGCF6PRH81o2kLWLWHgw= github.com/vishvananda/netns v0.0.5 h1:DfiHV+j8bA32MFM7bfEunvT8IAqQ/NzSJHtcmW5zdEY= github.com/vishvananda/netns v0.0.5/go.mod h1:SpkAiCQRtJ6TvvxPnOSyH3BMl6unz3xZlaprSwhNNJM= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= -go.datum.net/compute v0.8.0 h1:/v1lni/oO4KwthPeXhhS0+VFRZjWYbU6CXiF3RgCMzY= -go.datum.net/compute v0.8.0/go.mod h1:u7YIQX4+Wgts5XJwtvlz5NMs23g7+R/4TkAmv3DRycw= go.datum.net/compute v0.12.11 h1:bd/kcJMmuAP0pdLGV4OfOMKNg7GnYHVJwwJErYazBmE= go.datum.net/compute v0.12.11/go.mod h1:OydYUbFT+r76BB02bmBOltJR2nLEy080qofvdqk1LFU= -go.datum.net/network-services-operator v0.27.0 h1:LYCUjc6i0/f3c5YXSgisnbV8gB3R8emDwNfiYapdCAo= -go.datum.net/network-services-operator v0.27.0/go.mod h1:9nuuBWdrkdnIBMaWJsWM3j4CcbKJIF1GDuCmAjdpIHo= go.datum.net/network-services-operator v0.32.31 h1:OcUCPnAYj6ENiVyO4Wd24UeTvicJ/nSYkg2OQ5+quPk= go.datum.net/network-services-operator v0.32.31/go.mod h1:7LfOP7awc5FZ8zID7wZEN8rQ/7qxyO3u+WIoIIA44os= -go.miloapis.com/ipam v0.4.0 h1:U+mg3RMFXj0c2eQ0Lm15CI0bYBPFDqy8IsGLN8v/eGs= -go.miloapis.com/ipam v0.4.0/go.mod h1:Jj7xg4lJi9psE0+4PuOg/GQOG8rG13h112xYoM994rc= go.miloapis.com/ipam v0.5.1 h1:jb8jghj0jSOIDEPNo6b2gg2S4SWhNVfW4GDstb0zSKQ= go.miloapis.com/ipam v0.5.1/go.mod h1:Jj7xg4lJi9psE0+4PuOg/GQOG8rG13h112xYoM994rc= go.miloapis.com/locations v0.0.1 h1:voJKqBzyLX5x96M3+5y/ga7fgq9/+vypTizd+bBvqUY= @@ -294,8 +287,6 @@ gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc h1:rZHRz0ogq4Pid1IKtA/ivHi8XONVF5tqUQXA2TtgpkE= google.golang.org/genproto/googleapis/api v0.0.0-20260928230214-8a89bd6388cc/go.mod h1:K/n0XLazCJHjD73YuMyNJsN38BPPy1kfFnKxUXaThFU= -google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8 h1:0SXBqli5dnuCKMjI/X3FM226zEaEa0GUcSCJtN1vL2o= -google.golang.org/genproto/googleapis/rpc v0.0.0-20261005182115-fad411399dd8/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b h1:6WNWHZk91dp4gZ25642OALkuG1UVNZbwENJD19DnB8k= google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b/go.mod h1:OaIUM3+LpYcK2GXM4FTmhWoIq371Owdr+Cc7/BsYHHc= google.golang.org/grpc v1.84.0 h1:soMyaPJ8pAak5PIQ0DGBUir0XRo2fRoMqhNWMLlLxO0= @@ -327,8 +318,6 @@ sigs.k8s.io/controller-runtime v0.24.1 h1:miPEwrmirImAvgME1L9qebGHrOnGJoVmVdtOU9 sigs.k8s.io/controller-runtime v0.24.1/go.mod h1:vFkfY5fGt5xAC/sKb8IBFKgWPNKG9OUG29dR8Y2wImw= sigs.k8s.io/gateway-api v1.5.1 h1:RqVRIlkhLhUO8wOHKTLnTJA6o/1un4po4/6M1nRzdd0= sigs.k8s.io/gateway-api v1.5.1/go.mod h1:GvCETiaMAlLym5CovLxGjS0NysqFk3+Yuq3/rh6QL2o= -sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730 h1:IpInykpT6ceI+QxKBbEflcR5EXP7sU1kvOlxwZh5txg= -sigs.k8s.io/json v0.0.0-20250730193827-2d320260d730/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg= sigs.k8s.io/json v0.0.0-20260909141634-11ed52e25bc5 h1:k9jo7ED/fsfOcz3L1P20P5tjTNtsdQNwMOG2aKzHrl8= sigs.k8s.io/json v0.0.0-20260909141634-11ed52e25bc5/go.mod h1:mdzfpAEoE6DHQEN0uh9ZbOCuHbLK5wOm7dK4ctXE9Tg= sigs.k8s.io/knftables v0.0.18 h1:6Duvmu0s/HwGifKrtl6G3AyAPYlWiZqTgS8bkVMiyaE= diff --git a/internal/cloud/controller/bgpadvertisement_controller.go b/internal/cloud/controller/bgpadvertisement_controller.go index 1e2086aa..d38f2bbf 100644 --- a/internal/cloud/controller/bgpadvertisement_controller.go +++ b/internal/cloud/controller/bgpadvertisement_controller.go @@ -64,6 +64,9 @@ type BGPAdvertisementReconciler struct { func (r *BGPAdvertisementReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { var advertisement bgpv1alpha1.BGPAdvertisement if err := r.Get(ctx, req.NamespacedName, &advertisement); err != nil { + if apierrors.IsNotFound(err) { + return ctrl.Result{}, r.clearWithdrawnAdvertisement(ctx, req.NamespacedName) + } return ctrl.Result{}, client.IgnoreNotFound(err) } @@ -123,6 +126,72 @@ func (r *BGPAdvertisementReconciler) Reconcile(ctx context.Context, req ctrl.Req return ctrl.Result{}, nil } +// clearWithdrawnAdvertisement removes delivery status only when no tenant +// advertisement can certify the attachment, including after a node move. +func (r *BGPAdvertisementReconciler) clearWithdrawnAdvertisement(ctx context.Context, key types.NamespacedName) error { + vpc, attachmentID, ok := galactic.SplitAdvertisementName(key.Name) + if !ok || key.Namespace != galactic.SystemNamespace { + return nil + } + identity := galactic.AdvertisementName(vpc, attachmentID) + var advertisements bgpv1alpha1.BGPAdvertisementList + if err := r.List(ctx, &advertisements, client.InNamespace(key.Namespace), + client.MatchingFields{IndexBGPAdvertisementIdentity: identity}); err != nil { + return fmt.Errorf("list replacement advertisements: %w", err) + } + for i := range advertisements.Items { + advertisement := &advertisements.Items[i] + router, err := r.routerFor(ctx, advertisement.Namespace, advertisement.Spec.RouterRef.Name) + if err != nil { + return err + } + if router == nil || bgpv1alpha1.IsLegacyOwned(router) { + return nil + } + } + var attachments cloudv1alpha1.VPCAttachmentList + if err := r.List(ctx, &attachments, client.MatchingFields{IndexVPCAttachmentIdentity: identity}); err != nil { + return fmt.Errorf("list withdrawn attachments: %w", err) + } + for i := range attachments.Items { + attachment := &attachments.Items[i] + if attachment.Status.Node == "" || key.Name != identity+"-"+attachment.Status.Node { + continue + } + attachment.Status.Node = "" + attachment.Status.HostInterface = "" + attachment.Status.VRFInterface = "" + attachment.Status.PodSubnet = "" + programmed := metav1.Condition{ + Type: cloudv1alpha1.ConditionTypeProgrammed, Status: metav1.ConditionFalse, + Reason: "AdvertisementWithdrawn", Message: "the attachment has no network advertisement", + ObservedGeneration: attachment.Generation, + } + meta.SetStatusCondition(&attachment.Status.Conditions, programmed) + if err := r.projectOntoInterface(ctx, attachment, programmed); err != nil { + return err + } + if err := r.Status().Update(ctx, attachment); err != nil { + return fmt.Errorf("clear withdrawn attachment %s: %w", client.ObjectKeyFromObject(attachment), err) + } + } + return nil +} + +func advertisementForAttachment(_ context.Context, obj client.Object) []reconcile.Request { + attachment, ok := obj.(*cloudv1alpha1.VPCAttachment) + if !ok || attachment.Status.Node == "" { + return nil + } + identity := galactic.AdvertisementName(attachment.Status.VPC, attachment.Status.VPCAttachment) + if identity == "" { + return nil + } + return []reconcile.Request{{NamespacedName: types.NamespacedName{ + Namespace: galactic.SystemNamespace, Name: identity + "-" + attachment.Status.Node, + }}} +} + // projectOntoInterface closes the Programmed condition NSO deliberately leaves // for whoever realizes the interface. func (r *BGPAdvertisementReconciler) projectOntoInterface( @@ -279,6 +348,14 @@ func programmedCondition(advertisement *bgpv1alpha1.BGPAdvertisement) metav1.Con func (r *BGPAdvertisementReconciler) SetupWithManager(mgr ctrl.Manager) error { return ctrl.NewControllerManagedBy(mgr). For(&bgpv1alpha1.BGPAdvertisement{}). + Watches(&cloudv1alpha1.VPCAttachment{}, handler.EnqueueRequestsFromMapFunc(advertisementForAttachment), + builder.WithPredicates(predicate.Funcs{UpdateFunc: func(e event.UpdateEvent) bool { + oldAttachment, oldOK := e.ObjectOld.(*cloudv1alpha1.VPCAttachment) + newAttachment, newOK := e.ObjectNew.(*cloudv1alpha1.VPCAttachment) + return oldOK && newOK && (oldAttachment.Status.Node != newAttachment.Status.Node || + oldAttachment.Status.VPC != newAttachment.Status.VPC || + oldAttachment.Status.VPCAttachment != newAttachment.Status.VPCAttachment) + }})). Watches(&networkingv1alpha.NetworkInterface{}, handler.EnqueueRequestsFromMapFunc(r.advertisementForInterface), builder.WithPredicates(predicate.Funcs{ diff --git a/internal/cloud/controller/bgpadvertisement_ownership_test.go b/internal/cloud/controller/bgpadvertisement_ownership_test.go index bcb1a486..b53b333f 100644 --- a/internal/cloud/controller/bgpadvertisement_ownership_test.go +++ b/internal/cloud/controller/bgpadvertisement_ownership_test.go @@ -190,3 +190,59 @@ func TestReplayExcludesFabricAdvertisements(t *testing.T) { } } } + +func TestWithdrawnAdvertisementClearsDeliveryStatus(t *testing.T) { + f := newOwnershipFixture(t, false) + f.reconcile(t, f.tenantAdv) + if err := f.client.Delete(context.Background(), f.tenantAdv); err != nil { + t.Fatal(err) + } + // Replaying attachments also discovers withdrawals missed while the controller was offline. + var attachment cloudv1alpha1.VPCAttachment + if err := f.client.Get(context.Background(), client.ObjectKeyFromObject(f.attachment), &attachment); err != nil { + t.Fatal(err) + } + requests := advertisementForAttachment(context.Background(), &attachment) + if len(requests) != 1 || requests[0].Name != f.tenantAdv.Name { + t.Fatalf("withdrawal requests = %v", requests) + } + if _, err := f.reconciler.Reconcile(context.Background(), requests[0]); err != nil { + t.Fatal(err) + } + if err := f.client.Get(context.Background(), client.ObjectKeyFromObject(f.attachment), &attachment); err != nil { + t.Fatal(err) + } + condition := meta.FindStatusCondition(attachment.Status.Conditions, cloudv1alpha1.ConditionTypeProgrammed) + if condition == nil || condition.Status != metav1.ConditionFalse || attachment.Status.Node != "" || attachment.Status.HostInterface != "" { + t.Fatalf("withdrawn attachment status = %+v", attachment.Status) + } + var iface networkingv1alpha.NetworkInterface + if err := f.client.Get(context.Background(), client.ObjectKeyFromObject(f.iface), &iface); err != nil { + t.Fatal(err) + } + if !meta.IsStatusConditionFalse(iface.Status.Conditions, networkingv1alpha.NetworkInterfaceProgrammed) { + t.Fatalf("withdrawn interface status = %+v", iface.Status) + } +} + +func TestWithdrawalPreservesReplacementAdvertisement(t *testing.T) { + f := newOwnershipFixture(t, false) + f.reconcile(t, f.tenantAdv) + replacement := f.tenantAdv.DeepCopy() + replacement.Name += "-replacement" + replacement.ResourceVersion = "" + if err := f.client.Create(context.Background(), replacement); err != nil { + t.Fatal(err) + } + if err := f.client.Delete(context.Background(), f.tenantAdv); err != nil { + t.Fatal(err) + } + f.reconcile(t, f.tenantAdv) + var attachment cloudv1alpha1.VPCAttachment + if err := f.client.Get(context.Background(), client.ObjectKeyFromObject(f.attachment), &attachment); err != nil { + t.Fatal(err) + } + if !meta.IsStatusConditionTrue(attachment.Status.Conditions, cloudv1alpha1.ConditionTypeProgrammed) || attachment.Status.Node == "" { + t.Fatalf("replacement lost delivery status: %+v", attachment.Status) + } +} diff --git a/internal/cni/netns.go b/internal/cni/netns.go index 6b4f0487..a5a005d3 100644 --- a/internal/cni/netns.go +++ b/internal/cni/netns.go @@ -136,6 +136,15 @@ func addDefaultRouteIfMissing(handle *netlink.Handle, link netlink.Link, ifName return fmt.Errorf("default route on %q already points via %s, refusing to add conflicting route via %s", ifName, r.Gw, gateway) } + routes, err := handle.RouteList(nil, addrFamily(gateway)) + if err != nil { + return fmt.Errorf("list namespace routes: %w", err) + } + for _, route := range routes { + if isDefaultRouteDst(route.Dst) && route.LinkIndex != link.Attrs().Index { + return nil + } + } // The on-link flag: a VPC gateway sits outside the prefix allocated to the // attachment, for either family, so nothing on the interface tells the diff --git a/internal/cni/netns_test.go b/internal/cni/netns_test.go index 1d3749f8..bdbbc22b 100644 --- a/internal/cni/netns_test.go +++ b/internal/cni/netns_test.go @@ -5,6 +5,7 @@ package cni import ( + "errors" "fmt" "net" "os" @@ -59,7 +60,7 @@ func createTestNetnsWithDummy(t *testing.T) (netnsPath string, cleanup func()) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // best-effort cleanup + defer func() { _ = handle.Close() }() //nolint:errcheck // best-effort cleanup dummy := &netlink.Dummy{ LinkAttrs: netlink.LinkAttrs{Name: "test-dummy"}, @@ -84,7 +85,7 @@ func createTestNetnsWithDummy(t *testing.T) (netnsPath string, cleanup func()) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // best-effort cleanup + defer func() { _ = handle.Close() }() //nolint:errcheck // best-effort cleanup link, err := handle.LinkByName("test-dummy") if err != nil { return nil // already gone @@ -162,7 +163,7 @@ func TestCleanupContainerNetnsVeth(t *testing.T) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // best-effort cleanup + defer func() { _ = handle.Close() }() //nolint:errcheck // best-effort cleanup veth := &netlink.Veth{ LinkAttrs: netlink.LinkAttrs{Name: guestVethName}, @@ -193,7 +194,7 @@ func TestCleanupContainerNetnsVeth(t *testing.T) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // best-effort cleanup + defer func() { _ = handle.Close() }() //nolint:errcheck // best-effort cleanup for _, name := range []string{guestVethName, peerVethName} { if _, err := handle.LinkByName(name); err == nil { return fmt.Errorf("interface %q still exists after cleanup", name) @@ -234,7 +235,7 @@ func defaultRouteVia(t *testing.T, netnsPath string) net.IP { if err != nil { return err } - defer handle.Close() //nolint:errcheck // netlink cleanup on teardown + defer func() { _ = handle.Close() }() //nolint:errcheck // netlink cleanup on teardown link, err := handle.LinkByName("test-dummy") if err != nil { @@ -318,7 +319,7 @@ func TestConfigureInterfaceInNetnsAddAfterDel(t *testing.T) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // netlink cleanup on teardown + defer func() { _ = handle.Close() }() //nolint:errcheck // netlink cleanup on teardown link, err := handle.LinkByName("test-dummy") if err != nil { @@ -407,7 +408,7 @@ func TestFlushGuestNetnsConfigRemovesAddressAndRoute(t *testing.T) { if err != nil { return err } - defer handle.Close() //nolint:errcheck // netlink cleanup on teardown + defer func() { _ = handle.Close() }() //nolint:errcheck // netlink cleanup on teardown link, err := handle.LinkByName("test-dummy") if err != nil { @@ -476,3 +477,62 @@ func TestFlushGuestNetnsConfigThenRetryAddSucceeds(t *testing.T) { t.Fatalf("default route gateway = %v, want %v", gw, testGateway) } } + +func TestSecondaryInterfacePreservesPrimaryDefault(t *testing.T) { + netnsPath, cleanup := createTestNetnsWithDummy(t) + defer cleanup() + netns, err := ns.GetNS(netnsPath) + if err != nil { + t.Fatal(err) + } + defer func() { _ = netns.Close() }() + primaryGateway := net.ParseIP("fd00:99::1") + err = netns.Do(func(_ ns.NetNS) error { + handle, err := netlink.NewHandle() + if err != nil { + return err + } + defer func() { _ = handle.Close() }() + primary := &netlink.Dummy{LinkAttrs: netlink.LinkAttrs{Name: "primary"}} + if err := handle.LinkAdd(primary); err != nil { + return err + } + if err := handle.LinkSetUp(primary); err != nil { + return err + } + return handle.RouteAdd(&netlink.Route{ + Gw: primaryGateway, LinkIndex: primary.Attrs().Index, Flags: int(netlink.FLAG_ONLINK), + }) + }) + if err != nil { + t.Fatal(err) + } + for range 2 { + if err := configureInterfaceInNetns(netnsPath, "test-dummy", testIPNet, testGateway, nil, nil); err != nil { + t.Fatal(err) + } + } + if gateway := defaultRouteVia(t, netnsPath); gateway != nil { + t.Fatalf("secondary interface installed default route via %s", gateway) + } + err = netns.Do(func(_ ns.NetNS) error { + handle, err := netlink.NewHandle() + if err != nil { + return err + } + defer func() { _ = handle.Close() }() + routes, err := handle.RouteList(nil, netlink.FAMILY_V6) + if err != nil { + return err + } + for _, route := range routes { + if isDefaultRouteDst(route.Dst) && route.Gw.Equal(primaryGateway) { + return nil + } + } + return errors.New("primary default route is missing") + }) + if err != nil { + t.Fatal(err) + } +} diff --git a/internal/controller/serviceroute_access.go b/internal/controller/serviceroute_access.go index 8adf1b59..2ece513c 100644 --- a/internal/controller/serviceroute_access.go +++ b/internal/controller/serviceroute_access.go @@ -31,7 +31,13 @@ func (r *ServiceRoutePolicyReconciler) validateAuthorization(policy *networkv1al return nil } if policy.Spec.Authorization == nil { - return errors.New("frontend translation requires authorization.validUntil") + return errors.New("frontend translation requires access authorization") + } + if policy.Spec.Authorization.Persistent { + if !policy.Spec.Authorization.ValidUntil.IsZero() { + return errors.New("persistent access must omit validUntil") + } + return nil } deadline := policy.Spec.Authorization.ValidUntil.Time now := r.now() diff --git a/internal/controller/serviceroute_access_test.go b/internal/controller/serviceroute_access_test.go index ac0dc0be..c182215d 100644 --- a/internal/controller/serviceroute_access_test.go +++ b/internal/controller/serviceroute_access_test.go @@ -321,3 +321,59 @@ func TestConflictingPoliciesFailClosedAndRecover(t *testing.T) { }) } } + +func TestPersistentAuthorizationRequiresNoDeadline(t *testing.T) { + r := &ServiceRoutePolicyReconciler{} + p := &networkv1alpha1.ServiceRoutePolicy{Spec: networkv1alpha1.ServiceRoutePolicySpec{ + Frontend: &networkv1alpha1.ServiceRouteFrontend{}, + Authorization: &networkv1alpha1.ServiceRouteAuthorization{Persistent: true}, + }} + if err := r.validateAuthorization(p); err != nil { + t.Fatal(err) + } + p.Spec.Authorization.ValidUntil = metav1.NewTime(time.Now().Add(time.Minute)) + if err := r.validateAuthorization(p); err == nil { + t.Fatal("persistent authorization accepted a deadline") + } +} + +func TestPersistentPathSurvivesControllerOutageAndWithdraws(t *testing.T) { + r, p, _ := pscFixture(t) + ctx := context.Background() + now := time.Now().UTC() + r.Now = func() time.Time { return now } + p.Spec.Authorization = &networkv1alpha1.ServiceRouteAuthorization{Persistent: true} + if err := r.Update(ctx, p); err != nil { + t.Fatal(err) + } + key := types.NamespacedName{Namespace: p.Namespace, Name: p.Name} + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + now = now.Add(time.Hour) + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + if len(r.Applied) == 0 { + t.Fatal("persistent path requires authorization renewal") + } + for _, intents := range r.Applied { + for _, intent := range intents { + if !intent.AuthorizationValidUntil.IsZero() { + t.Fatal("persistent path has a dataplane deadline") + } + } + } + if err := r.Get(ctx, key, p); err != nil { + t.Fatal(err) + } + if err := r.Delete(ctx, p); err != nil { + t.Fatal(err) + } + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + if len(r.Applied) != 0 { + t.Fatal("withdrawn policy retained forwarding") + } +} diff --git a/internal/controller/serviceroute_controller.go b/internal/controller/serviceroute_controller.go index 163c7d0d..0b3cd313 100644 --- a/internal/controller/serviceroute_controller.go +++ b/internal/controller/serviceroute_controller.go @@ -188,7 +188,7 @@ func (r *ServiceRoutePolicyReconciler) Reconcile(ctx context.Context, req ctrl.R all = append(all, &attachments.Items[i]) } if policy.Spec.Frontend != nil { - inputDigest = serviceProgrammingDigest(policy, endpoint, all, vpcIdentity) + inputDigest = serviceProgrammingDigest(policy, endpoint, all, vpcIdentity, r.NodeName) } conflicts, conflictErr := r.conflictingPolicies(ctx, all) if conflictErr != nil { diff --git a/internal/controller/serviceroute_programming.go b/internal/controller/serviceroute_programming.go index 6ea3b29d..8fc1440c 100644 --- a/internal/controller/serviceroute_programming.go +++ b/internal/controller/serviceroute_programming.go @@ -6,6 +6,7 @@ package controller import ( "context" "errors" + "slices" "sort" "time" @@ -24,7 +25,8 @@ func (r *ServiceRoutePolicyReconciler) setNodeProgramming( ctx context.Context, policy *networkv1alpha1.ServiceRoutePolicy, digest string, ready bool, ) error { deadline := r.now().Add(serviceProgrammingAckDuration) - if policy.Spec.Authorization != nil && policy.Spec.Authorization.ValidUntil.Before(&metav1.Time{Time: deadline}) { + if policy.Spec.Authorization != nil && !policy.Spec.Authorization.Persistent && + policy.Spec.Authorization.ValidUntil.Before(&metav1.Time{Time: deadline}) { deadline = policy.Spec.Authorization.ValidUntil.Time } report := networkv1alpha1.ServiceRouteNodeStatus{ @@ -51,7 +53,7 @@ func (r *ServiceRoutePolicyReconciler) setNodeProgramming( found = true } } - if !found && !ready { + if !found && !ready && digest == "" { return nil } if !found { @@ -65,10 +67,10 @@ func (r *ServiceRoutePolicyReconciler) setNodeProgramming( } func serviceProgrammingDigest(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alpha1.ServiceEndpoint, - attachments []*cloudv1alpha1.VPCAttachment, vpcIdentity string, + attachments []*cloudv1alpha1.VPCAttachment, vpcIdentity, nodeName string, ) string { plan, err := serviceroute.ProgrammingPlan(policy, endpoint, attachments, vpcIdentity) - if err != nil { + if err != nil || !slices.Contains(plan.Nodes, nodeName) { return "" } return plan.InputDigest diff --git a/internal/controller/serviceroute_programming_test.go b/internal/controller/serviceroute_programming_test.go index 98ce4ece..21c74585 100644 --- a/internal/controller/serviceroute_programming_test.go +++ b/internal/controller/serviceroute_programming_test.go @@ -159,3 +159,46 @@ func TestDisablingFrontendWithdrawsNodeAcknowledgment(t *testing.T) { t.Fatal("disabled node retained programming proof") } } + +func TestInitialProgrammingFailureReportsUnreadyNode(t *testing.T) { + r, p, _ := pscFixture(t) + r.Programmer = &retryRouteProgrammer{failApply: 1} + ctx := context.Background() + key := types.NamespacedName{Namespace: p.Namespace, Name: p.Name} + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err == nil { + t.Fatal("expected programming failure") + } + current := &networkv1alpha1.ServiceRoutePolicy{} + if err := r.Get(ctx, key, current); err != nil { + t.Fatal(err) + } + if len(current.Status.Nodes) != 1 || current.Status.Nodes[0].Ready || current.Status.Nodes[0].InputDigest == "" { + t.Fatalf("missing initial failure report: %+v", current.Status.Nodes) + } + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + if err := r.Get(ctx, key, current); err != nil { + t.Fatal(err) + } + if len(current.Status.Nodes) != 1 || !current.Status.Nodes[0].Ready { + t.Fatalf("missing recovery report: %+v", current.Status.Nodes) + } +} + +func TestUninvolvedNodeDoesNotPublishProgrammingReport(t *testing.T) { + r, p, _ := pscFixture(t) + r.NodeName = "uninvolved" + ctx := context.Background() + key := types.NamespacedName{Namespace: p.Namespace, Name: p.Name} + if _, err := r.Reconcile(ctx, ctrl.Request{NamespacedName: key}); err != nil { + t.Fatal(err) + } + current := &networkv1alpha1.ServiceRoutePolicy{} + if err := r.Get(ctx, key, current); err != nil { + t.Fatal(err) + } + if len(current.Status.Nodes) != 0 { + t.Fatalf("uninvolved node published a report: %+v", current.Status.Nodes) + } +} diff --git a/internal/gc/datapath_repair.go b/internal/gc/datapath_repair.go index d02c73aa..ff4d67b1 100644 --- a/internal/gc/datapath_repair.go +++ b/internal/gc/datapath_repair.go @@ -557,7 +557,8 @@ func (c routeCache) get(tableID uint32) ([]netlink.Route, error) { // localAddressing recovers what CNI ADD registered as link's local prefixes and // gateways from the kernel. The guest prefixes are the pod-subnet routes the // master plugin installs in the VRF's table out link: unicast, not -// kernel-generated, with no gateway and not link-scoped. The master plugin +// kernel-generated, and not link-scoped. Delegated veth prefixes use their +// guest address as an on-link next hop. The master plugin // installs one only for a family with a gateway, so a family without one is // missing here; discoverAttachments adds it from the BGPAdvertisement. The // gateways are link's own addresses that are not link-local, each also a host @@ -610,9 +611,10 @@ func setGateway(gw *attachreg.Gateways, ip net.IP) { } // isPodSubnetRoute reports whether r is a pod-subnet route out ifindex, as -// opposed to a kernel-generated route or a termination route with a gateway or -// link scope. +// opposed to a kernel-generated route or a termination route. A delegated +// prefix can use its first address as an on-link guest next hop. func isPodSubnetRoute(r *netlink.Route, ifindex int) bool { - return r.Dst != nil && r.LinkIndex == ifindex && r.Gw == nil && len(r.MultiPath) == 0 && + return r.Dst != nil && r.LinkIndex == ifindex && (r.Gw == nil || + (r.Gw.Equal(r.Dst.IP) && r.Flags&int(netlink.FLAG_ONLINK) != 0)) && len(r.MultiPath) == 0 && r.Type == unix.RTN_UNICAST && r.Protocol != unix.RTPROT_KERNEL && r.Scope != netlink.SCOPE_LINK } diff --git a/internal/gc/datapath_repair_test.go b/internal/gc/datapath_repair_test.go index 0c53941f..cc8949ba 100644 --- a/internal/gc/datapath_repair_test.go +++ b/internal/gc/datapath_repair_test.go @@ -1054,3 +1054,23 @@ func TestRepairAttachmentDatapath_NodeIdentity(t *testing.T) { } }) } + +func TestRepairAttachmentDatapathDelegatedGuestRoute(t *testing.T) { + f := newRepairFixture(t) + vrf := f.addVRF("G0000000jUV", 10, 7) + f.addVRFInstance(crdnames.BGPVRFInstanceName("jU", repairNode), 0x21) + f.addHostLink(attachreg.InterfaceTypeVeth, "G0000000jUabcH", 11, vrf) + f.routes[7] = append(f.routes[7], netlink.Route{ + Dst: mustCIDR(t, "fd20:30:ff01::/96"), LinkIndex: 11, + Type: unix.RTN_UNICAST, Protocol: unix.RTPROT_BOOT, + Gw: net.ParseIP("fd20:30:ff01::"), Flags: int(netlink.FLAG_ONLINK), + }) + result, err := f.run() + if err != nil { + t.Fatal(err) + } + if result.Rebuilt.LocalEgressRoutes != 1 { + t.Fatalf("rebuilt %d routes, want 1", result.Rebuilt.LocalEgressRoutes) + } + wantPassThrough(t, f.mem, 7, "fd20:30:ff01::/96") +} diff --git a/internal/hostgw/hostgw.go b/internal/hostgw/hostgw.go index 139f7257..4ffc3975 100644 --- a/internal/hostgw/hostgw.go +++ b/internal/hostgw/hostgw.go @@ -225,6 +225,10 @@ func installPodSubnetRoute(hostLink netlink.Link, subnet *net.IPNet, family, tab LinkIndex: hostLink.Attrs().Index, Table: tableID, } + if ones, bits := subnet.Mask.Size(); hostLink.Type() == "veth" && bits == 128 && ones < bits { + desiredRoute.Gw = subnet.IP + desiredRoute.Flags = int(netlink.FLAG_ONLINK) + } existingRoutes, err := netlink.RouteListFiltered( family, @@ -241,6 +245,14 @@ func installPodSubnetRoute(hostLink netlink.Link, subnet *net.IPNet, family, tab continue } if r.Dst.String() == desiredRoute.Dst.String() { + if r.LinkIndex == desiredRoute.LinkIndex && r.Gw == nil && desiredRoute.Gw != nil && + isPodSubnetRouteCandidate(&r) && len(r.MultiPath) == 0 { + if err := netlink.RouteReplace(desiredRoute); err != nil { + return fmt.Errorf("route delegated subnet through guest address: %w", err) + } + alreadyPresent = true + continue + } if routeConflicts(&r, desiredRoute) { return fmt.Errorf( "existing route %v to %s conflicts with desired route %v", diff --git a/internal/hostgw/hostgw_test.go b/internal/hostgw/hostgw_test.go index 115dbf5d..8fa5763b 100644 --- a/internal/hostgw/hostgw_test.go +++ b/internal/hostgw/hostgw_test.go @@ -303,3 +303,32 @@ func addDummyLink(t *testing.T, name string) netlink.Link { }) return link } + +func TestDelegatedVethSubnetUsesGuestNextHop(t *testing.T) { + requireRoot(t) + link := &netlink.Veth{LinkAttrs: netlink.LinkAttrs{Name: "hgwdns0"}, PeerName: "hgwdns1"} + if err := netlink.LinkAdd(link); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = netlink.LinkDel(link) }) + if err := netlink.LinkSetUp(link); err != nil { + t.Fatal(err) + } + subnet := mustParseCIDR(t, "fd20:0:e::2:0:0/96") + const tableID = 251 + if err := netlink.RouteAdd(&netlink.Route{Dst: subnet, LinkIndex: link.Attrs().Index, Table: tableID}); err != nil { + t.Fatal(err) + } + for range 2 { + if err := installPodSubnetRoute(link, subnet, netlink.FAMILY_V6, tableID, true); err != nil { + t.Fatal(err) + } + } + routes, err := netlink.RouteListFiltered(netlink.FAMILY_V6, &netlink.Route{Table: tableID}, netlink.RT_FILTER_TABLE) + if err != nil { + t.Fatal(err) + } + if len(routes) != 1 || !routes[0].Gw.Equal(subnet.IP) || routes[0].Flags&int(netlink.FLAG_ONLINK) == 0 { + t.Fatalf("delegated route must use the guest next hop: %+v", routes) + } +} diff --git a/internal/plumbing/ebpf/prog/doc.go b/internal/plumbing/ebpf/prog/doc.go index b25a5eb4..fd910248 100644 --- a/internal/plumbing/ebpf/prog/doc.go +++ b/internal/plumbing/ebpf/prog/doc.go @@ -40,4 +40,4 @@ package prog // -cc is deliberately omitted: bpf2go defaults to $BPF2GO_CC or "clang", so CI // can pin an exact version without this directive changing. // -//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cflags "-O2 -g -Wall -idirafter /usr/include/x86_64-linux-gnu -idirafter /usr/include/aarch64-linux-gnu" -target bpfel,bpfeb -type locator_value -type function_value -type vrf_value -type ifindex_vrf_value -type nptv6_value -type vip_xlat_key -type vip_xlat_value -type egress_route_key -type egress_route_value -type service_route_key -type service_route_value -type service_access_key -type service_access_value -type service_deny_key -type service_deny_value -type service_reverse_key -type service_reverse_value -type service_remote_grant_key -type service_remote_grant_value -type service_policy_state_value -type public_uplink_value -type egress_shard_group_value -type egress_shard_member -type egress_shard_pin_key -type egress_shard_pin_value -type egress_shard_counter -type mss_clamp_value -type tenant_gw_value -type vpc_attribution_value -type vrf_direction_counter -type vrf_traffic_value -type tp_state_value -type tp_iface_value -type tp_iface_addr_key -type tp_binding_key -type tp_entry -type tp_header -type tp_counter -type tp_fallback_key Usid usid.c +//go:generate go run github.com/cilium/ebpf/cmd/bpf2go -cflags "-O2 -g -Wall -idirafter /usr/include/x86_64-linux-gnu -idirafter /usr/include/aarch64-linux-gnu" -target bpfel,bpfeb -type locator_value -type function_value -type vrf_value -type ifindex_vrf_value -type nptv6_value -type vip_xlat_key -type vip_xlat_value -type egress_route_key -type egress_route_value -type service_route_key -type service_route_value -type service_access_key -type service_access_value -type service_deny_key -type service_deny_value -type service_reverse_key -type service_reverse_value -type service_remote_grant_key -type service_remote_grant_value -type service_peer_key -type service_peer_value -type service_policy_state_value -type public_uplink_value -type egress_shard_group_value -type egress_shard_member -type egress_shard_pin_key -type egress_shard_pin_value -type egress_shard_counter -type mss_clamp_value -type tenant_gw_value -type vpc_attribution_value -type vrf_direction_counter -type vrf_traffic_value -type tp_state_value -type tp_iface_value -type tp_iface_addr_key -type tp_binding_key -type tp_entry -type tp_header -type tp_counter -type tp_fallback_key Usid usid.c diff --git a/internal/plumbing/ebpf/prog/usid.c b/internal/plumbing/ebpf/prog/usid.c index bb963dd8..4dbfb005 100644 --- a/internal/plumbing/ebpf/prog/usid.c +++ b/internal/plumbing/ebpf/prog/usid.c @@ -1454,6 +1454,27 @@ struct { __type(value, __u64); } attachment_identity_table SEC(".maps"); +struct service_peer_key { + __u32 ifindex; + __u8 family; + __u8 pad[3]; +}; + +struct service_peer_value { + __u64 token; + __u8 address[16]; +}; + +// One next hop per attachment and family delivers virtual service addresses +// through the guest's neighbor entry without resolving each virtual address. +struct { + __uint(type, BPF_MAP_TYPE_HASH); + __uint(map_flags, BPF_F_NO_PREALLOC); + __uint(max_entries, 65536); + __type(key, struct service_peer_key); + __type(value, struct service_peer_value); +} service_peer_table SEC(".maps"); + struct { __uint(type, BPF_MAP_TYPE_HASH); __uint(map_flags, BPF_F_NO_PREALLOC); @@ -4537,12 +4558,22 @@ static USID_ALWAYS_INLINE long service_redirect(struct __sk_buff *skb, if (service_decrement_hop(skb, scratch)) return TC_ACT_SHOT; __builtin_memset(&scratch->neigh, 0, sizeof(scratch->neigh)); + struct service_peer_key key = {.ifindex = target_ifindex, .family = scratch->meta_family}; + struct service_peer_value *peer = bpf_map_lookup_elem(&service_peer_table, &key); + if (peer && !service_identity_matches(target_ifindex, peer->token)) + return TC_ACT_SHOT; if (scratch->meta_family == USID_EGRESS_ROUTE_FAMILY_INET6) { scratch->neigh.nh_family = USID_AF_INET6; - __builtin_memcpy(scratch->neigh.ipv6_nh, scratch->dest_addr, 16); + if (peer) + __builtin_memcpy(scratch->neigh.ipv6_nh, peer->address, 16); + else + __builtin_memcpy(scratch->neigh.ipv6_nh, scratch->dest_addr, 16); } else { scratch->neigh.nh_family = USID_AF_INET; - __builtin_memcpy(&scratch->neigh.ipv4_nh, scratch->dest_addr, 4); + if (peer) + __builtin_memcpy(&scratch->neigh.ipv4_nh, peer->address, 4); + else + __builtin_memcpy(&scratch->neigh.ipv4_nh, scratch->dest_addr, 4); } return bpf_redirect_neigh(target_ifindex, &scratch->neigh, sizeof(scratch->neigh), 0); } diff --git a/internal/plumbing/ebpf/prog/usid_test.go b/internal/plumbing/ebpf/prog/usid_test.go index 9d6d7238..3c040cce 100644 --- a/internal/plumbing/ebpf/prog/usid_test.go +++ b/internal/plumbing/ebpf/prog/usid_test.go @@ -1567,6 +1567,21 @@ func TestUsidEgress_ServiceRouteAllowsAuthorizedPort(t *testing.T) { if got := sumPerCPU(t, objs.DropReasons, 40); got != 0 { t.Errorf("drop_reasons[service_route_denied] = %d, want 0", got) } + peerKey := UsidServicePeerKey{Ifindex: loopback, Family: egressRouteFamilyINET6} + peer := UsidServicePeerValue{Token: 99, Address: netip.MustParseAddr("fd20::4").As16()} + if err := objs.ServicePeerTable.Put(peerKey, peer); err != nil { + t.Fatal(err) + } + if ret, _, err := objs.UsidServiceEgress.Test(pkt); err != nil || ret != tcActShot { + t.Fatalf("stale attachment next hop verdict = %d, err = %v; want drop", ret, err) + } + peer.Token = 1 + if err := objs.ServicePeerTable.Put(peerKey, peer); err != nil { + t.Fatal(err) + } + if ret, _, err := objs.UsidServiceEgress.Test(pkt); err != nil || ret != tcActRedirect { + t.Fatalf("current attachment next hop verdict = %d, err = %v; want redirect", ret, err) + } // Simulate deletion and immediate kernel ifindex reuse. The stale route and // access keys still match ifindex 1, but their old incarnation token must // not authorize the replacement interface. diff --git a/internal/plumbing/ebpf/serviceroutemap/table.go b/internal/plumbing/ebpf/serviceroutemap/table.go index a9d72916..8c1dfff4 100644 --- a/internal/plumbing/ebpf/serviceroutemap/table.go +++ b/internal/plumbing/ebpf/serviceroutemap/table.go @@ -27,6 +27,7 @@ const ( attachmentIdentityMapName = "attachment_identity_table" servicePolicyStateMapName = "service_policy_state_table" serviceDenyMapName = "service_deny_table" + servicePeerMapName = "service_peer_table" familyIPv6 = uint8(0) familyIPv4 = uint8(1) @@ -196,6 +197,7 @@ type Tables struct { identities usidmap.Table policyState usidmap.Table deny usidmap.Table + peers usidmap.Table leaseClockFn func() (time.Time, uint64, error) } @@ -248,12 +250,19 @@ func OpenPinned(pinDir string) (*Tables, []*ebpf.Map, error) { } return nil, nil, fmt.Errorf("serviceroutemap: open %s: %w", serviceDenyMapName, err) } + peers, err := ebpf.LoadPinnedMap(filepath.Join(pinDir, servicePeerMapName), nil) + if err != nil { + for _, opened := range []*ebpf.Map{routes, access, reverse, remoteGrants, identities, policyState, deny} { + _ = opened.Close() + } + return nil, nil, fmt.Errorf("serviceroutemap: open %s: %w", servicePeerMapName, err) + } return New( usidmap.KernelTable{Map: routes}, usidmap.KernelTable{Map: access}, usidmap.KernelTable{Map: reverse}, usidmap.KernelTable{Map: remoteGrants}, usidmap.KernelTable{Map: identities}, - usidmap.KernelTable{Map: policyState}, usidmap.KernelTable{Map: deny}, + usidmap.KernelTable{Map: policyState}, usidmap.KernelTable{Map: deny}, usidmap.KernelTable{Map: peers}, ), - []*ebpf.Map{routes, access, reverse, remoteGrants, identities, policyState, deny}, nil + []*ebpf.Map{routes, access, reverse, remoteGrants, identities, policyState, deny, peers}, nil } // New wraps map implementations. Production uses OpenPinned; tests use fakes. @@ -271,9 +280,71 @@ func New(routes, access, reverse usidmap.Table, remoteGrants ...usidmap.Table) * if len(remoteGrants) > 3 { t.deny = remoteGrants[3] } + if len(remoteGrants) > 4 { + t.peers = remoteGrants[4] + } return t } +type servicePeerKey struct { + Ifindex uint32 + Family uint8 + Pad [3]uint8 +} + +type servicePeerValue struct { + Token uint64 + Address [16]uint8 +} + +// SetPeer records an attachment's next hop, fenced by its current identity. +func (t *Tables) SetPeer(ifindex uint32, token uint64, address net.IP) error { + if address == nil { + return nil + } + if t.peers == nil { + return errors.New("serviceroutemap: service peer table unavailable") + } + key := servicePeerKey{Ifindex: ifindex} + if address.To4() != nil { + key.Family = familyIPv4 + } + value := servicePeerValue{Token: token} + copy(value.Address[:], addressBytes(address)) + if err := t.peers.Put(key, value); err != nil { + return fmt.Errorf("serviceroutemap: register attachment next hop: %w", err) + } + return nil +} + +func (t *Tables) sweepPeers() error { + if t.peers == nil { + return nil + } + iterator := t.peers.Iterate() + var key servicePeerKey + var value servicePeerValue + var stale []servicePeerKey + for iterator.Next(&key, &value) { + token, err := t.IdentityToken(key.Ifindex) + if err != nil && !errors.Is(err, ebpf.ErrKeyNotExist) { + return err + } + if err != nil || token != value.Token { + stale = append(stale, key) + } + } + if err := iterator.Err(); err != nil { + return fmt.Errorf("serviceroutemap: iterate attachment next hops: %w", err) + } + for _, key := range stale { + if err := t.peers.Delete(key); err != nil && !errors.Is(err, ebpf.ErrKeyNotExist) { + return fmt.Errorf("serviceroutemap: remove attachment next hop: %w", err) + } + } + return nil +} + // SetPolicyEnabled gates all service authorization while the controller is // rebuilding a complete cache-synchronized generation. func (t *Tables) SetPolicyEnabled(enabled bool) error { @@ -519,6 +590,7 @@ func (t *Tables) SweepPolicy(snapshot PolicySnapshot) error { deleteErrs = append(deleteErrs, fmt.Errorf("serviceroutemap: sweep deny catalog: %w", err)) } } + deleteErrs = append(deleteErrs, t.sweepPeers()) return errors.Join(deleteErrs...) } diff --git a/internal/plumbing/ebpf/serviceroutemap/table_test.go b/internal/plumbing/ebpf/serviceroutemap/table_test.go index ab56518a..dda3f429 100644 --- a/internal/plumbing/ebpf/serviceroutemap/table_test.go +++ b/internal/plumbing/ebpf/serviceroutemap/table_test.go @@ -422,3 +422,39 @@ func TestAuthorizationLeasePinnedReplayAndRenewal(t *testing.T) { }) } } + +func TestPeerNextHopFollowsAttachmentIdentity(t *testing.T) { + identities := &memoryTable{} + peers := &memoryTable{} + tables := &Tables{identities: identities, peers: peers} + if err := identities.Put(uint32(42), uint64(7)); err != nil { + t.Fatal(err) + } + for _, address := range []string{"fd20::4", "192.0.2.4"} { + if err := tables.SetPeer(42, 7, net.ParseIP(address)); err != nil { + t.Fatal(err) + } + } + if err := tables.sweepPeers(); err != nil { + t.Fatal(err) + } + if len(peers.entries) != 2 { + t.Fatalf("live peer entries = %d, want 2", len(peers.entries)) + } + var v6 servicePeerValue + if err := peers.Lookup(servicePeerKey{Ifindex: 42}, &v6); err != nil { + t.Fatal(err) + } + if !net.IP(v6.Address[:]).Equal(net.ParseIP("fd20::4")) { + t.Fatalf("IPv6 next hop = %x", v6.Address) + } + if err := identities.Put(uint32(42), uint64(8)); err != nil { + t.Fatal(err) + } + if err := tables.sweepPeers(); err != nil { + t.Fatal(err) + } + if len(peers.entries) != 0 { + t.Fatal("recycled attachment retained next hops") + } +} diff --git a/internal/serviceroute/cleanup_test.go b/internal/serviceroute/cleanup_test.go index dc70b703..48414c86 100644 --- a/internal/serviceroute/cleanup_test.go +++ b/internal/serviceroute/cleanup_test.go @@ -19,9 +19,9 @@ func TestCleanupDrainsFailedApplyRollbackBeforeFinalize(t *testing.T) { routes := &recordingTable{putErr: applyFailure} access := &recordingTable{deleteErr: rollbackFailure} programmer := testRouteProgrammer(routes, access, &recordingTable{}) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } programmer.targetIndexFn = func(name string) (uint32, error) { switch name { case testConsumerDeviceName: diff --git a/internal/serviceroute/dataplane.go b/internal/serviceroute/dataplane.go index 5d978879..39866830 100644 --- a/internal/serviceroute/dataplane.go +++ b/internal/serviceroute/dataplane.go @@ -111,7 +111,7 @@ type EBPFRouteProgrammer struct { mu sync.Mutex tables *serviceroutemap.Tables mapHandles []*ebpf.Map - mapIDs [7]ebpf.MapID + mapIDs [8]ebpf.MapID mapIDsValid bool routeRefs map[string]routeState accessRefs map[accessKey]accessState @@ -128,8 +128,8 @@ type EBPFRouteProgrammer struct { // These hooks keep map replacement and interface recreation tests // deterministic without requiring a privileged kernel eBPF setup. - openMapSetFn func(string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) - readMapIDsFn func(string) ([7]ebpf.MapID, error) + openMapSetFn func(string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) + readMapIDsFn func(string) ([8]ebpf.MapID, error) targetIndexFn func(string) (uint32, error) } @@ -388,6 +388,7 @@ var serviceRouteMapNames = [...]string{ "attachment_identity_table", "service_policy_state_table", "service_deny_table", + "service_peer_table", } func (p *EBPFRouteProgrammer) mapSetChanged() (bool, error) { @@ -401,32 +402,32 @@ func (p *EBPFRouteProgrammer) mapSetChanged() (bool, error) { return ids != p.mapIDs, nil } -func (p *EBPFRouteProgrammer) openMapSet(pinDir string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) { +func (p *EBPFRouteProgrammer) openMapSet(pinDir string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) { if p.openMapSetFn != nil { return p.openMapSetFn(pinDir) } tables, handles, err := serviceroutemap.OpenPinned(pinDir) if err != nil { - return nil, nil, [7]ebpf.MapID{}, err + return nil, nil, [8]ebpf.MapID{}, err } - var ids [7]ebpf.MapID + var ids [8]ebpf.MapID for index, handle := range handles { ids[index], err = kernelMapID(handle) if err != nil { for _, opened := range handles { _ = opened.Close() } - return nil, nil, [7]ebpf.MapID{}, err + return nil, nil, [8]ebpf.MapID{}, err } } return tables, handles, ids, nil } -func (p *EBPFRouteProgrammer) readMapIDs(pinDir string) ([7]ebpf.MapID, error) { +func (p *EBPFRouteProgrammer) readMapIDs(pinDir string) ([8]ebpf.MapID, error) { if p.readMapIDsFn != nil { return p.readMapIDsFn(pinDir) } - var ids [7]ebpf.MapID + var ids [8]ebpf.MapID for index, name := range serviceRouteMapNames { current, err := ebpf.LoadPinnedMap(filepath.Join(pinDir, name), nil) if err != nil { @@ -459,14 +460,25 @@ func (p *EBPFRouteProgrammer) closeHandles() { } p.mapHandles = nil p.tables = nil - p.mapIDs = [7]ebpf.MapID{} + p.mapIDs = [8]ebpf.MapID{} p.mapIDsValid = false } +// Missing links withdraw their entries while preserving intent for recreation. +// Other lookup errors cannot prove that an interface has disappeared. +func (p *EBPFRouteProgrammer) recoverableEntries(intent RouteIntent) ([]appliedEntry, error) { + entries, err := p.entries(intent) + var missing netlink.LinkNotFoundError + if errors.As(err, &missing) { + return []appliedEntry{}, nil + } + return entries, err +} + func (p *EBPFRouteProgrammer) rebuildDesired() error { for _, key := range sortedIntentKeys(p.desiredIntents) { for _, intent := range p.desiredIntents[key] { - entries, err := p.entries(intent) + entries, err := p.recoverableEntries(intent) if err != nil { return fmt.Errorf("re-derive service route state %s: %w", key, err) } @@ -494,7 +506,7 @@ func (p *EBPFRouteProgrammer) repairDesired() error { continue } for index, intent := range intents { - fresh, err := p.entries(intent) + fresh, err := p.recoverableEntries(intent) if err != nil { return fmt.Errorf("re-derive service route state %s: %w", key, err) } @@ -657,6 +669,9 @@ func (p *EBPFRouteProgrammer) entries(intent RouteIntent) ([]appliedEntry, error if err != nil { return nil, err } + if err := p.tables.SetPeer(serviceIfindex, producerToken, intent.ServiceNextHop); err != nil { + return nil, err + } // Producer-side source marker for fragments, whose missing transport // tuple cannot be matched against an individual grant. entries = append(entries, appliedEntry{access: &accessRef{ @@ -688,6 +703,9 @@ func (p *EBPFRouteProgrammer) entries(intent RouteIntent) ([]appliedEntry, error if err != nil { return nil, err } + if err := p.tables.SetPeer(consumerIfindex, consumerToken, intent.ConsumerNextHop); err != nil { + return nil, err + } // A protocol/port-zero marker makes malformed or non-TCP/UDP traffic to a // configured service fail closed without conflating distinct real tuples. entries = append(entries, appliedEntry{access: &accessRef{ @@ -715,6 +733,9 @@ func (p *EBPFRouteProgrammer) entries(intent RouteIntent) ([]appliedEntry, error if err != nil { return nil, err } + if err := p.tables.SetPeer(targetIfindex, producerToken, intent.ServiceNextHop); err != nil { + return nil, err + } // Local replies need the same address-only producer marker. Its // refcount is independent from the consumer marker because ifindex is // part of the key. @@ -1014,6 +1035,7 @@ func intentKey(intent RouteIntent) string { fmt.Fprintf(&builder, "%s|%s|%s|%s|%s|%s|%s|%s|%s|%x", intent.Attachment, intent.ProducerAttachment, intent.Kind, intent.Service, intent.Frontend, intent.ConsumerDevice, intent.ServiceDevice, intent.ServiceSID, intent.ConsumerSID, intent.GrantID) + fmt.Fprintf(&builder, "|peers:%s,%s", intent.ConsumerNextHop, intent.ServiceNextHop) fmt.Fprintf(&builder, "|lease:%d", authorizationUnixNano(intent.AuthorizationValidUntil)) for _, port := range intent.Ports { fmt.Fprintf(&builder, "|%s:%d", port.Protocol, port.Port) @@ -1035,6 +1057,8 @@ func cloneRouteIntent(intent RouteIntent) RouteIntent { } cloned.ServiceSID = append(net.IP(nil), intent.ServiceSID...) cloned.ConsumerSID = append(net.IP(nil), intent.ConsumerSID...) + cloned.ConsumerNextHop = append(net.IP(nil), intent.ConsumerNextHop...) + cloned.ServiceNextHop = append(net.IP(nil), intent.ServiceNextHop...) cloned.Ports = append([]api.ServiceRouteProtocolPort(nil), intent.Ports...) return cloned } diff --git a/internal/serviceroute/dataplane_test.go b/internal/serviceroute/dataplane_test.go index 522e9290..318dd000 100644 --- a/internal/serviceroute/dataplane_test.go +++ b/internal/serviceroute/dataplane_test.go @@ -16,6 +16,7 @@ import ( "time" "github.com/cilium/ebpf" + "github.com/vishvananda/netlink" api "go.datum.net/galactic/api/network/v1alpha1" "go.datum.net/galactic/internal/plumbing/ebpf/serviceroutemap" @@ -264,10 +265,10 @@ func TestInitializePreservesPinnedState(t *testing.T) { reverse := &recordingTable{iterateCount: 1} grants := &recordingTable{iterateCount: 1} programmer := &EBPFRouteProgrammer{ - openMapSetFn: func(string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) { + openMapSetFn: func(string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) { return serviceroutemap.New(routes, access, reverse, grants, &recordingTable{identityTokens: map[uint32]uint64{}}, &recordingTable{}, &recordingTable{}), nil, - [7]ebpf.MapID{1, 2, 3, 4, 5, 6}, nil + [8]ebpf.MapID{1, 2, 3, 4, 5, 6}, nil }, } @@ -282,7 +283,7 @@ func TestInitializePreservesPinnedState(t *testing.T) { func TestInitializeDoesNotDisableFinalizedPolicy(t *testing.T) { policyState := &recordingTable{} - ids := [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + ids := [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer := &EBPFRouteProgrammer{ tables: serviceroutemap.New( &recordingTable{}, &recordingTable{}, &recordingTable{}, &recordingTable{}, @@ -297,7 +298,7 @@ func TestInitializeDoesNotDisableFinalizedPolicy(t *testing.T) { appliedRefs: make(map[string][][]appliedEntry), desiredIntents: make(map[string][]RouteIntent), pendingRollbacks: make(map[string][]appliedEntry), - readMapIDsFn: func(string) ([7]ebpf.MapID, error) { return ids, nil }, + readMapIDsFn: func(string) ([8]ebpf.MapID, error) { return ids, nil }, } if err := programmer.Initialize(); err != nil { t.Fatalf("first Initialize: %v", err) @@ -316,9 +317,9 @@ func TestInitializeDoesNotDisableFinalizedPolicy(t *testing.T) { func TestInitializeRepairsRecreatedInterface(t *testing.T) { routes, access, grants := &recordingTable{}, &recordingTable{}, &recordingTable{} programmer := testRouteProgrammer(routes, access, grants) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } consumerIndex := uint32(10) programmer.targetIndexFn = func(name string) (uint32, error) { switch name { @@ -354,9 +355,9 @@ func TestInitializeRepairsRecreatedInterface(t *testing.T) { func TestInitializeRepairsSharedRefsAsCompleteGeneration(t *testing.T) { routes, access, grants := &recordingTable{}, &recordingTable{}, &recordingTable{} programmer := testRouteProgrammer(routes, access, grants) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } consumerIndex := uint32(10) programmer.targetIndexFn = func(name string) (uint32, error) { switch name { @@ -406,9 +407,9 @@ func TestInitializeRepairsSharedRefsAsCompleteGeneration(t *testing.T) { func TestInitializeRetriesIncompleteGenerationRebuild(t *testing.T) { routes, access, grants := &recordingTable{}, &recordingTable{}, &recordingTable{} programmer := testRouteProgrammer(routes, access, grants) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } consumerIndex := uint32(10) programmer.targetIndexFn = func(name string) (uint32, error) { if name == testConsumerDeviceName { @@ -443,9 +444,9 @@ func TestInitializeRetriesIncompleteGenerationRebuild(t *testing.T) { func TestRemoveRetryClearsSyncFailure(t *testing.T) { routes := &recordingTable{deleteErr: errors.New("transient delete failure")} programmer := testRouteProgrammer(routes, &recordingTable{}, &recordingTable{}) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } programmer.targetIndexFn = func(name string) (uint32, error) { if name == testConsumerDeviceName { return 10, nil @@ -473,12 +474,12 @@ func TestRemoveRetryClearsSyncFailure(t *testing.T) { func TestMapReplacementClearsStaleMutationFailure(t *testing.T) { routes := &recordingTable{deleteErr: errors.New("transient delete failure")} programmer := testRouteProgrammer(routes, &recordingTable{}, &recordingTable{}) - oldIDs := [7]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} - newIDs := [7]ebpf.MapID{11, 12, 13, 14, 15, 16, 17} + oldIDs := [8]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} + newIDs := [8]ebpf.MapID{11, 12, 13, 14, 15, 16, 17} currentIDs := oldIDs programmer.mapIDs = oldIDs programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return currentIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return currentIDs, nil } programmer.targetIndexFn = func(name string) (uint32, error) { if name == testConsumerDeviceName { return 10, nil @@ -502,7 +503,7 @@ func TestMapReplacementClearsStaleMutationFailure(t *testing.T) { newRoutes := &recordingTable{} newPolicyState := &recordingTable{} currentIDs = newIDs - programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) { + programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) { return serviceroutemap.New(newRoutes, &recordingTable{}, &recordingTable{}, &recordingTable{}, &recordingTable{identityTokens: map[uint32]uint64{10: 100, 20: 200}}, newPolicyState, &recordingTable{}), nil, newIDs, nil @@ -525,10 +526,10 @@ func TestMapReplacementClearsStaleMutationFailure(t *testing.T) { func TestInitializeReopensAllMapsWhenGrantMapChanges(t *testing.T) { oldRoutes, oldAccess, oldGrants := &recordingTable{}, &recordingTable{}, &recordingTable{} programmer := testRouteProgrammer(oldRoutes, oldAccess, oldGrants) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} programmer.mapIDsValid = true currentIDs := programmer.mapIDs - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return currentIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return currentIDs, nil } programmer.targetIndexFn = func(name string) (uint32, error) { if name == testConsumerDeviceName { return 10, nil @@ -543,7 +544,7 @@ func TestInitializeReopensAllMapsWhenGrantMapChanges(t *testing.T) { newRoutes, newAccess, newGrants := &recordingTable{}, &recordingTable{}, &recordingTable{} openCalls := 0 currentIDs[3] = 99 - programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) { + programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) { openCalls++ return serviceroutemap.New( newRoutes, newAccess, &recordingTable{}, newGrants, @@ -682,9 +683,9 @@ func leasedTestProgrammer() (*EBPFRouteProgrammer, *persistedPolicyTable, *persi programmer := testRouteProgrammer(&recordingTable{}, &recordingTable{}, &recordingTable{}) programmer.tables = serviceroutemap.New(routes, &persistedPolicyTable{}, &persistedPolicyTable{}, grants, &recordingTable{identityTokens: map[uint32]uint64{10: 100, 20: 200, 42: 420}}, gate, &persistedPolicyTable{}) - programmer.mapIDs = [7]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6, 7} programmer.mapIDsValid = true - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return programmer.mapIDs, nil } + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } programmer.targetIndexFn = func(name string) (uint32, error) { switch name { case testConsumerDeviceName: @@ -815,9 +816,9 @@ func TestMapReplayPrunesExpiredAuthorizationAndPreservesStartupGate(t *testing.T t.Fatal(err) } } - newIDs := [7]ebpf.MapID{11, 12, 13, 14, 15, 16, 17} - programmer.readMapIDsFn = func(string) ([7]ebpf.MapID, error) { return newIDs, nil } - programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [7]ebpf.MapID, error) { + newIDs := [8]ebpf.MapID{11, 12, 13, 14, 15, 16, 17} + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return newIDs, nil } + programmer.openMapSetFn = func(string) (*serviceroutemap.Tables, []*ebpf.Map, [8]ebpf.MapID, error) { return serviceroutemap.New(newRoutes, &persistedPolicyTable{}, &persistedPolicyTable{}, &persistedPolicyTable{}, &recordingTable{identityTokens: map[uint32]uint64{10: 100, 20: 200, 42: 420}}, newGate, &persistedPolicyTable{}), nil, newIDs, nil @@ -842,3 +843,115 @@ func TestMapReplayPrunesExpiredAuthorizationAndPreservesStartupGate(t *testing.T }) } } + +func TestPersistentDataplaneHasNoExpiry(t *testing.T) { + programmer, _, _, _ := leasedTestProgrammer() + intent := localTestIntent() + if err := programmer.Apply(intent); err != nil { + t.Fatal(err) + } + if expired := programmer.expiredDesiredKeys(time.Now().Add(24 * time.Hour)); len(expired) != 0 { + t.Fatalf("persistent intents expired: %v", expired) + } + if err := programmer.Remove(intent); err != nil { + t.Fatal(err) + } +} + +func TestInitializeWithdrawsMissingInterfaceAndRecovers(t *testing.T) { + programmer := testRouteProgrammer(&recordingTable{}, &recordingTable{}, &recordingTable{}) + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDsValid = true + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } + missing := false + programmer.targetIndexFn = func(name string) (uint32, error) { + if name == testConsumerDeviceName && missing { + return 0, netlink.LinkNotFoundError{} + } + if name == testConsumerDeviceName { + return 10, nil + } + return 20, nil + } + intent := localTestIntent() + if err := programmer.Apply(intent); err != nil { + t.Fatal(err) + } + missing = true + if err := programmer.Initialize(); err != nil { + t.Fatalf("withdraw missing link: %v", err) + } + if len(programmer.routeRefs) != 0 || len(programmer.accessRefs) != 0 { + t.Fatal("missing interface retained programmed routes or access") + } + if len(programmer.desiredIntents[intentKey(intent)]) != 1 { + t.Fatal("lost recoverable intent") + } + missing = false + if err := programmer.Initialize(); err != nil { + t.Fatalf("recover recreated link: %v", err) + } + if len(programmer.routeRefs) == 0 { + t.Fatal("recreated interface did not recover routes") + } + missing = true + if err := programmer.Remove(intent); err != nil { + t.Fatalf("remove missing link: %v", err) + } + if len(programmer.desiredIntents) != 0 { + t.Fatal("removed intent retained for replay") + } +} + +func TestMissingInterfaceDoesNotBlockOtherRoutes(t *testing.T) { + programmer := testRouteProgrammer(&recordingTable{}, &recordingTable{}, &recordingTable{}) + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDsValid = true + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } + missing := false + programmer.targetIndexFn = func(name string) (uint32, error) { + switch name { + case testConsumerDeviceName: + if missing { + return 0, netlink.LinkNotFoundError{} + } + return 10, nil + case testOtherConsumerDevice: + return 11, nil + default: + return 20, nil + } + } + intent := localTestIntent() + if err := programmer.Apply(intent); err != nil { + t.Fatal(err) + } + missing = true + other := localTestIntent() + other.ConsumerDevice = testOtherConsumerDevice + if err := programmer.Apply(other); err != nil { + t.Fatalf("unrelated route blocked: %v", err) + } + if len(programmer.routeRefs) != 1 { + t.Fatalf("routes = %d, want only live route", len(programmer.routeRefs)) + } + if err := programmer.Remove(intent); err != nil { + t.Fatal(err) + } + if len(programmer.appliedRefs[intentKey(other)]) != 1 { + t.Fatal("unrelated route lost during cleanup") + } +} + +func TestRepairPropagatesInterfaceLookupErrors(t *testing.T) { + programmer := testRouteProgrammer(&recordingTable{}, &recordingTable{}, &recordingTable{}) + programmer.mapIDs = [8]ebpf.MapID{1, 2, 3, 4, 5, 6} + programmer.mapIDsValid = true + programmer.readMapIDsFn = func(string) ([8]ebpf.MapID, error) { return programmer.mapIDs, nil } + lookupErr := errors.New("netlink unavailable") + programmer.targetIndexFn = func(string) (uint32, error) { return 0, lookupErr } + programmer.desiredIntents["test"] = []RouteIntent{localTestIntent()} + if err := programmer.Initialize(); !errors.Is(err, lookupErr) { + t.Fatalf("Initialize = %v, want lookup error", err) + } +} diff --git a/internal/serviceroute/planner.go b/internal/serviceroute/planner.go index 6a84f808..98f24322 100644 --- a/internal/serviceroute/planner.go +++ b/internal/serviceroute/planner.go @@ -41,6 +41,8 @@ type RouteIntent struct { Frontend *net.IPNet ConsumerDevice string ServiceDevice string + ConsumerNextHop net.IP + ServiceNextHop net.IP ServiceSID net.IP ConsumerSID net.IP GrantID ServiceGrantID @@ -143,6 +145,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentLocal, Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ConsumerDevice: consumer.Status.HostInterface, ServiceDevice: producer.Status.HostInterface, Ports: clonePorts(ports), + ConsumerNextHop: attachmentNextHop(consumer, service.IP), ServiceNextHop: attachmentNextHop(producer, service.IP), }) } continue @@ -164,6 +167,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentRemoteConsumer, Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ConsumerDevice: consumer.Status.HostInterface, ServiceSID: cloneIP(serviceSID), GrantID: grantID, Ports: clonePorts(ports), + ConsumerNextHop: attachmentNextHop(consumer, service.IP), }) } if producer.Status.Node == localNode { @@ -176,6 +180,7 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph Attachment: consumerKey, ProducerAttachment: producerKey, Kind: RouteIntentRemoteProducer, Service: cloneIPNet(service), Frontend: cloneIPNet(frontend), ServiceDevice: producer.Status.HostInterface, ConsumerSID: cloneIP(consumerSID), GrantID: grantID, Ports: clonePorts(ports), + ServiceNextHop: attachmentNextHop(producer, service.IP), }) } } @@ -189,6 +194,15 @@ func compile(policy *networkv1alpha1.ServiceRoutePolicy, endpoint *networkv1alph return intents, nil } +// attachmentNextHop returns the guest address for a routed attachment subnet. +func attachmentNextHop(attachment *cloudv1alpha1.VPCAttachment, service net.IP) net.IP { + ip, _, err := net.ParseCIDR(attachment.Status.PodSubnet) + if err != nil || (ip.To4() == nil) != (service.To4() == nil) { + return nil + } + return ip +} + func validateProducerTopology(endpoint *networkv1alpha1.ServiceEndpoint, producers []*cloudv1alpha1.VPCAttachment, ) error { @@ -371,7 +385,7 @@ func consumerMatchesVPC(policy *networkv1alpha1.ServiceRoutePolicy, } func setIntentAuthorization(policy *networkv1alpha1.ServiceRoutePolicy, intents []RouteIntent) { - if policy.Spec.Frontend == nil || policy.Spec.Authorization == nil { + if policy.Spec.Frontend == nil || policy.Spec.Authorization == nil || policy.Spec.Authorization.Persistent { return } for i := range intents { diff --git a/internal/serviceroute/planner_test.go b/internal/serviceroute/planner_test.go index 8ed6d711..8da1e8ed 100644 --- a/internal/serviceroute/planner_test.go +++ b/internal/serviceroute/planner_test.go @@ -458,3 +458,19 @@ func TestServiceGrantGenerationCompatibility(t *testing.T) { }) } } + +func TestAttachmentNextHopUsesGuestAddress(t *testing.T) { + attachment := &cloudv1alpha1.VPCAttachment{} + attachment.Status.PodSubnet = "fd20::4:0:0/96" + got := attachmentNextHop(attachment, net.ParseIP("fd70::53")) + if !got.Equal(net.ParseIP("fd20::4:0:0")) { + t.Fatalf("next hop = %v", got) + } + if got := attachmentNextHop(attachment, net.ParseIP("192.0.2.53")); got != nil { + t.Fatalf("wrong-family next hop = %v", got) + } + attachment.Status.PodSubnet = "" + if got := attachmentNextHop(attachment, net.ParseIP("fd70::53")); got != nil { + t.Fatalf("missing subnet next hop = %v", got) + } +}