From 532b967296c516e267bc5f7b1570161a54eb4822 Mon Sep 17 00:00:00 2001 From: Scot Wells Date: Thu, 27 Aug 2026 17:56:35 -0500 Subject: [PATCH] feat: carry a VPC pod to the edge only when served Every VPC pod's EndpointSlice is federated to every edge cluster, so a cell publishes one tenant's pod addresses and SRv6 SIDs onto edges that serve other tenants and other locations. The staging lab cell carries 178 slices this way. Nothing filters them, because the two facts that decide whether an edge needs a pod, the NetworkService selecting it and the HTTPProxy naming that service, live in a project control plane a cell cannot read. EdgeReachability records the answer where both planes meet. The control plane resolves, per project namespace, the workload addresses behind a proxy, and writes them to the federation hub. The write-back reads the record from the hub it already publishes into and carries a slice only when the record names one of its addresses. The record answers per namespace rather than per service so a reader can tell "nothing is served here" from "the control plane has not answered". Silence keeps a pod published: a route withdrawn under a pod that is still serving black-holes live traffic, while a route left up for a pod nothing sends to costs a table entry. Key changes: - Add EdgeReachability, a hub-only record of a project's served addresses - Add EdgeReachabilityReconciler, resolving proxies, services and the interfaces they select in each project control plane - Filter the VPC EndpointSlice write-back on the record, collecting a copy that stops serving - Resync published slices every minute, since nothing in a cell watches the hub and a pod put behind a proxy has no local event to act on --- Taskfile.test-infra.yml | 1 + api/v1alpha/edgereachability_types.go | 57 ++++ api/v1alpha/groupversion_info.go | 2 + api/v1alpha/zz_generated.deepcopy.go | 78 +++++ ...king.datumapis.com_edgereachabilities.yaml | 76 +++++ config/crd/downstream/kustomization.yaml | 6 + config/rbac/role.yaml | 1 + config/rbac_downstream/role.yaml | 12 + internal/cmd/manager/manager.go | 8 + internal/cmd/manager/manager_test.go | 1 + .../controller/edgereachability_controller.go | 248 +++++++++++++++ .../edgereachability_controller_test.go | 292 ++++++++++++++++++ .../networkinterfaceclaim_controller.go | 4 +- .../controller/vpcendpointslice_writeback.go | 98 +++++- .../vpcendpointslice_writeback_test.go | 108 +++++++ 15 files changed, 986 insertions(+), 6 deletions(-) create mode 100644 api/v1alpha/edgereachability_types.go create mode 100644 config/crd/bases/networking.datumapis.com_edgereachabilities.yaml create mode 100644 internal/controller/edgereachability_controller.go create mode 100644 internal/controller/edgereachability_controller_test.go diff --git a/Taskfile.test-infra.yml b/Taskfile.test-infra.yml index 85fe3ed1..fb4f7ca5 100644 --- a/Taskfile.test-infra.yml +++ b/Taskfile.test-infra.yml @@ -414,6 +414,7 @@ tasks: desc: "Install the NSO CRDs the replicator mirrors into the downstream cluster (the Gateway-API/EG CRDs come from eg-crds)." cmds: - kubectl --context {{.DOWNSTREAM_CTX}} apply -f config/crd/bases/networking.datumapis.com_connectors.yaml + - kubectl --context {{.DOWNSTREAM_CTX}} apply -f config/crd/bases/networking.datumapis.com_edgereachabilities.yaml - kubectl --context {{.DOWNSTREAM_CTX}} apply -f config/crd/bases/networking.datumapis.com_httpproxies.yaml - kubectl --context {{.DOWNSTREAM_CTX}} apply -f config/crd/bases/networking.datumapis.com_trafficprotectionpolicies.yaml - kubectl --context {{.DOWNSTREAM_CTX}} apply -f config/crd/bases/networking.datumapis.com_servinglocations.yaml diff --git a/api/v1alpha/edgereachability_types.go b/api/v1alpha/edgereachability_types.go new file mode 100644 index 00000000..add4cd5b --- /dev/null +++ b/api/v1alpha/edgereachability_types.go @@ -0,0 +1,57 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package v1alpha + +import ( + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" +) + +// EdgeReachabilityName is the name of the single record a namespace holds. +// One record answers for the whole namespace, so a reader gets its answer with +// a get rather than a list it has to decide is complete. +const EdgeReachabilityName = "default" + +// EdgeReachabilitySpec is the set of workload addresses in one project +// namespace that an edge is expected to reach. +type EdgeReachabilitySpec struct { + // addresses are the workload addresses currently behind a proxy, one entry + // per address, with no prefix length. An empty list is a real answer: it + // says the project publishes nothing, which is different from no record at + // all. + // + // +kubebuilder:validation:Optional + // +kubebuilder:validation:MaxItems=8192 + Addresses []string `json:"addresses,omitempty"` +} + +// +kubebuilder:object:root=true + +// EdgeReachability records which of a project's workload addresses are behind +// an HTTPProxy, so the platform carries a workload's location to the edges that +// serve it and stops carrying it everywhere else. +// +// It is written by the control plane onto the federation hub and read by the +// cells publishing into it. No consumer creates or edits one, and nothing in a +// project control plane holds one. +// +// Absence of the record means the control plane has not answered for this +// namespace yet, and a reader must keep publishing rather than treat silence as +// a withdrawal. An empty list is the answer that withdraws. +// +kubebuilder:printcolumn:name="Addresses",type=integer,JSONPath=".spec.addresses.length()" +// +kubebuilder:printcolumn:name="Age",type="date",JSONPath=".metadata.creationTimestamp" +type EdgeReachability struct { + metav1.TypeMeta `json:",inline"` + metav1.ObjectMeta `json:"metadata,omitempty"` + + // +kubebuilder:validation:Optional + Spec EdgeReachabilitySpec `json:"spec,omitempty"` +} + +// +kubebuilder:object:root=true + +// EdgeReachabilityList contains a list of EdgeReachability. +type EdgeReachabilityList struct { + metav1.TypeMeta `json:",inline"` + metav1.ListMeta `json:"metadata,omitempty"` + Items []EdgeReachability `json:"items"` +} diff --git a/api/v1alpha/groupversion_info.go b/api/v1alpha/groupversion_info.go index 0d6572bd..1483afb9 100644 --- a/api/v1alpha/groupversion_info.go +++ b/api/v1alpha/groupversion_info.go @@ -26,6 +26,8 @@ func addKnownTypes(scheme *runtime.Scheme) error { scheme.AddKnownTypes(GroupVersion, &Domain{}, &DomainList{}, + &EdgeReachability{}, + &EdgeReachabilityList{}, &HTTPProxy{}, &HTTPProxyList{}, &Location{}, diff --git a/api/v1alpha/zz_generated.deepcopy.go b/api/v1alpha/zz_generated.deepcopy.go index eae1923e..8ae5b3ea 100644 --- a/api/v1alpha/zz_generated.deepcopy.go +++ b/api/v1alpha/zz_generated.deepcopy.go @@ -300,6 +300,84 @@ func (in *DomainVerificationStatus) DeepCopy() *DomainVerificationStatus { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EdgeReachability) DeepCopyInto(out *EdgeReachability) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ObjectMeta.DeepCopyInto(&out.ObjectMeta) + in.Spec.DeepCopyInto(&out.Spec) +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EdgeReachability. +func (in *EdgeReachability) DeepCopy() *EdgeReachability { + if in == nil { + return nil + } + out := new(EdgeReachability) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EdgeReachability) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EdgeReachabilityList) DeepCopyInto(out *EdgeReachabilityList) { + *out = *in + out.TypeMeta = in.TypeMeta + in.ListMeta.DeepCopyInto(&out.ListMeta) + if in.Items != nil { + in, out := &in.Items, &out.Items + *out = make([]EdgeReachability, len(*in)) + for i := range *in { + (*in)[i].DeepCopyInto(&(*out)[i]) + } + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EdgeReachabilityList. +func (in *EdgeReachabilityList) DeepCopy() *EdgeReachabilityList { + if in == nil { + return nil + } + out := new(EdgeReachabilityList) + in.DeepCopyInto(out) + return out +} + +// DeepCopyObject is an autogenerated deepcopy function, copying the receiver, creating a new runtime.Object. +func (in *EdgeReachabilityList) DeepCopyObject() runtime.Object { + if c := in.DeepCopy(); c != nil { + return c + } + return nil +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *EdgeReachabilitySpec) DeepCopyInto(out *EdgeReachabilitySpec) { + *out = *in + if in.Addresses != nil { + in, out := &in.Addresses, &out.Addresses + *out = make([]string, len(*in)) + copy(*out, *in) + } +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new EdgeReachabilitySpec. +func (in *EdgeReachabilitySpec) DeepCopy() *EdgeReachabilitySpec { + if in == nil { + return nil + } + out := new(EdgeReachabilitySpec) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *GCPLocationProvider) DeepCopyInto(out *GCPLocationProvider) { *out = *in diff --git a/config/crd/bases/networking.datumapis.com_edgereachabilities.yaml b/config/crd/bases/networking.datumapis.com_edgereachabilities.yaml new file mode 100644 index 00000000..a855f682 --- /dev/null +++ b/config/crd/bases/networking.datumapis.com_edgereachabilities.yaml @@ -0,0 +1,76 @@ +--- +apiVersion: apiextensions.k8s.io/v1 +kind: CustomResourceDefinition +metadata: + annotations: + controller-gen.kubebuilder.io/version: v0.16.4 + name: edgereachabilities.networking.datumapis.com +spec: + group: networking.datumapis.com + names: + kind: EdgeReachability + listKind: EdgeReachabilityList + plural: edgereachabilities + singular: edgereachability + scope: Namespaced + versions: + - additionalPrinterColumns: + - jsonPath: .spec.addresses.length() + name: Addresses + type: integer + - jsonPath: .metadata.creationTimestamp + name: Age + type: date + name: v1alpha + schema: + openAPIV3Schema: + description: |- + EdgeReachability records which of a project's workload addresses are behind + an HTTPProxy, so the platform carries a workload's location to the edges that + serve it and stops carrying it everywhere else. + + It is written by the control plane onto the federation hub and read by the + cells publishing into it. No consumer creates or edits one, and nothing in a + project control plane holds one. + + Absence of the record means the control plane has not answered for this + namespace yet, and a reader must keep publishing rather than treat silence as + a withdrawal. An empty list is the answer that withdraws. + properties: + apiVersion: + description: |- + APIVersion defines the versioned schema of this representation of an object. + Servers should convert recognized schemas to the latest internal value, and + may reject unrecognized values. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#resources + type: string + kind: + description: |- + Kind is a string value representing the REST resource this object represents. + Servers may infer this from the endpoint the client submits requests to. + Cannot be updated. + In CamelCase. + More info: https://git.k8s.io/community/contributors/devel/sig-architecture/api-conventions.md#types-kinds + type: string + metadata: + type: object + spec: + description: |- + EdgeReachabilitySpec is the set of workload addresses in one project + namespace that an edge is expected to reach. + properties: + addresses: + description: |- + addresses are the workload addresses currently behind a proxy, one entry + per address, with no prefix length. An empty list is a real answer: it + says the project publishes nothing, which is different from no record at + all. + items: + type: string + maxItems: 8192 + type: array + type: object + type: object + served: true + storage: true + subresources: {} diff --git a/config/crd/downstream/kustomization.yaml b/config/crd/downstream/kustomization.yaml index 54a7b9a0..029554fa 100644 --- a/config/crd/downstream/kustomization.yaml +++ b/config/crd/downstream/kustomization.yaml @@ -7,6 +7,11 @@ # TrafficProtectionPolicy, HTTPProxy and Connector downstream, and the # extension server's cache reads them locally. # +# Edge reachability: the control plane records, on the hub, which of a project's +# workload addresses are behind a proxy, and the cells publishing into that +# namespace read it to decide what to carry. It stops at the hub; no policy +# carries it on. +# # Network presence: a consumer writes a NetworkBinding on the hub, the control # plane manager turns every binding for a (network, location) pair into one # shared NetworkContext in the project control plane, the replicator mirrors @@ -26,6 +31,7 @@ # NetworkContext, Subnet, and ServingLocation on the local cluster. resources: # Gateway data plane + - ../bases/networking.datumapis.com_edgereachabilities.yaml - ../bases/networking.datumapis.com_trafficprotectionpolicies.yaml - ../bases/networking.datumapis.com_httpproxies.yaml - ../bases/networking.datumapis.com_connectors.yaml diff --git a/config/rbac/role.yaml b/config/rbac/role.yaml index ece18574..edce3a98 100644 --- a/config/rbac/role.yaml +++ b/config/rbac/role.yaml @@ -242,6 +242,7 @@ rules: - connectoradvertisements - connectors - domains + - edgereachabilities - httpproxies - networkbindings - networkcontexts diff --git a/config/rbac_downstream/role.yaml b/config/rbac_downstream/role.yaml index c1880d33..7837a4c1 100644 --- a/config/rbac_downstream/role.yaml +++ b/config/rbac_downstream/role.yaml @@ -182,3 +182,15 @@ rules: - envoypatchpolicies/status verbs: - get +- apiGroups: + - networking.datumapis.com + resources: + - edgereachabilities + verbs: + - create + - delete + - get + - list + - patch + - update + - watch diff --git a/internal/cmd/manager/manager.go b/internal/cmd/manager/manager.go index bc685ad6..2a349eda 100644 --- a/internal/cmd/manager/manager.go +++ b/internal/cmd/manager/manager.go @@ -649,6 +649,14 @@ func controllerRegistrations( {"networkservice", true, func() error { return (&controller.NetworkServiceReconciler{}).SetupWithManager(mgr) }}, + // Which workloads are behind a proxy is only answerable in a project + // control plane, and only a cell can act on it. The record this writes to + // the hub is what carries the answer between them. + {"edgereachability", true, func() error { + return (&controller.EdgeReachabilityReconciler{ + DownstreamCluster: deps.downstreamCluster, + }).SetupWithManager(mgr) + }}, {"subnet", true, func() error { return (&controller.SubnetReconciler{}).SetupWithManager(mgr) }}, diff --git a/internal/cmd/manager/manager_test.go b/internal/cmd/manager/manager_test.go index 964b8830..fa04e95f 100644 --- a/internal/cmd/manager/manager_test.go +++ b/internal/cmd/manager/manager_test.go @@ -23,6 +23,7 @@ var reconcilerControllerNames = map[string]string{ "ConnectorAdvertisementReconciler": "connectoradvertisement", "ConnectorReconciler": "connector", "DomainReconciler": "domain", + "EdgeReachabilityReconciler": "edgereachability", "GatewayClassReconciler": "gatewayclass", "GatewayDownstreamCertificateSolverReconciler": "downstream-certificate-solver", "GatewayDownstreamGCReconciler": "gateway_downstream_resources", diff --git a/internal/controller/edgereachability_controller.go b/internal/controller/edgereachability_controller.go new file mode 100644 index 00000000..feaf119e --- /dev/null +++ b/internal/controller/edgereachability_controller.go @@ -0,0 +1,248 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package controller + +import ( + "context" + "errors" + "fmt" + "slices" + + discoveryv1 "k8s.io/api/discovery/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/cluster" + "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/log" + mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder" + mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" + "sigs.k8s.io/multicluster-runtime/pkg/multicluster" + mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile" + + networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" + "go.datum.net/network-services-operator/internal/downstreamclient" +) + +// EdgeReachabilityReconciler records, per project namespace, the workload +// addresses that are behind an HTTPProxy, so that what a cell carries to the +// edge is the workloads the edge actually serves rather than every workload the +// project runs. +// +// It runs where both halves of the question are answerable: a project control +// plane holds the proxies, the services, and the interfaces a service selects. +// The answer is written to the federation hub, which is the only plane a cell +// can also read. +// +// The record is written per namespace rather than per service so that a reader +// can tell "the control plane says nothing is published here" from "the control +// plane has not answered". Those are the same absence when the answer is a set +// of objects, and they call for opposite behaviour at the edge. +type EdgeReachabilityReconciler struct { + // DownstreamCluster is the federation hub the records are written to. + DownstreamCluster cluster.Cluster + + mgr mcmanager.Manager +} + +// +kubebuilder:rbac:groups=networking.datumapis.com,resources=edgereachabilities,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=networking.datumapis.com,resources=httpproxies,verbs=get;list;watch +// +kubebuilder:rbac:groups=networking.datumapis.com,resources=networkservices,verbs=get;list;watch +// +kubebuilder:rbac:groups=networking.datumapis.com,resources=networkinterfaces,verbs=get;list;watch + +func (r *EdgeReachabilityReconciler) Reconcile(ctx context.Context, req mcreconcile.Request) (ctrl.Result, error) { + cl, err := r.mgr.GetCluster(ctx, req.ClusterName) + if err != nil { + return ctrl.Result{}, err + } + + return ctrl.Result{}, r.record(ctx, string(req.ClusterName), cl.GetClient(), req.Namespace) +} + +func (r *EdgeReachabilityReconciler) record( + ctx context.Context, + clusterName string, + cl client.Client, + namespace string, +) error { + logger := log.FromContext(ctx) + + addresses, err := r.publishedAddresses(ctx, cl, namespace) + if err != nil { + return err + } + + strategy := downstreamclient.NewMappedNamespaceResourceStrategy(clusterName, cl, r.DownstreamCluster.GetClient()) + + hubNamespace, err := strategy.GetDownstreamNamespaceNameForUpstreamNamespace(ctx, namespace) + if err != nil { + if apierrors.IsNotFound(err) { + return nil + } + return fmt.Errorf("failed resolving the hub namespace for %q: %w", namespace, err) + } + + record := &networkingv1alpha.EdgeReachability{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: hubNamespace, + Name: networkingv1alpha.EdgeReachabilityName, + }, + } + + _, err = controllerutil.CreateOrUpdate(ctx, r.DownstreamCluster.GetClient(), record, func() error { + record.Spec.Addresses = addresses + return nil + }) + if err != nil { + // The hub namespace is made by the federation that carries a project's + // work out. Nothing here should invent one, and a namespace that is not + // there yet holds nothing to withdraw. + if apierrors.IsNotFound(err) { + logger.Info("the hub has no namespace to record edge reachability in yet", + "namespace", namespace) + return nil + } + return fmt.Errorf("failed recording edge reachability for %q: %w", namespace, err) + } + + return nil +} + +// publishedAddresses is every workload address a proxy in this namespace sends +// traffic to. A proxy naming a service resolves through the same selector the +// service's membership is resolved through, so the answer here and the endpoints +// the proxy is programmed with cannot disagree. +func (r *EdgeReachabilityReconciler) publishedAddresses( + ctx context.Context, + cl client.Client, + namespace string, +) ([]string, error) { + logger := log.FromContext(ctx) + + var proxies networkingv1alpha.HTTPProxyList + if err := cl.List(ctx, &proxies, client.InNamespace(namespace)); err != nil { + return nil, fmt.Errorf("failed listing http proxies in %q: %w", namespace, err) + } + + seen := map[string]struct{}{} + var addresses []string + add := func(address string) { + if address == "" { + return + } + if _, ok := seen[address]; ok { + return + } + seen[address] = struct{}{} + addresses = append(addresses, address) + } + + services := map[string]struct{}{} + instanceBackends := map[string]struct{}{} + + for i := range proxies.Items { + proxy := &proxies.Items[i] + if !proxy.DeletionTimestamp.IsZero() { + continue + } + for _, rule := range proxy.Spec.Rules { + for _, backend := range rule.Backends { + if backend.NetworkService != nil { + services[backend.NetworkService.Name] = struct{}{} + } + if backend.Instance != nil { + instanceBackends[backend.Instance.Name] = struct{}{} + } + } + } + } + + for name := range services { + var service networkingv1alpha.NetworkService + key := client.ObjectKey{Namespace: namespace, Name: name} + if err := cl.Get(ctx, key, &service); err != nil { + if apierrors.IsNotFound(err) { + continue + } + return nil, fmt.Errorf("failed getting network service %q: %w", name, err) + } + + selector, err := metav1.LabelSelectorAsSelector(&service.Spec.NetworkInterfaces.Selector) + if err != nil { + logger.Info("not recording the members of a network service whose selector cannot be evaluated", + "namespace", namespace, jsonKeyName, name) + continue + } + + members, err := matchingInterfaces(ctx, cl, namespace, selector) + if err != nil { + return nil, fmt.Errorf("failed resolving the members of network service %q: %w", name, err) + } + + for i := range members { + for _, address := range interfaceBackhaulAddresses(&members[i]) { + add(address) + } + } + } + + for name := range instanceBackends { + var slice discoveryv1.EndpointSlice + key := client.ObjectKey{Namespace: namespace, Name: name} + if err := cl.Get(ctx, key, &slice); err != nil { + if apierrors.IsNotFound(err) { + continue + } + return nil, fmt.Errorf("failed getting instance backend endpointslice %q: %w", name, err) + } + for _, endpoint := range slice.Endpoints { + for _, address := range endpoint.Addresses { + add(address) + } + } + } + + slices.Sort(addresses) + return addresses, nil +} + +// SetupWithManager sets up the controller with the Manager. +func (r *EdgeReachabilityReconciler) SetupWithManager(mgr mcmanager.Manager) error { + if r.DownstreamCluster == nil { + return errors.New("a downstream cluster is required") + } + + r.mgr = mgr + + return mcbuilder.ControllerManagedBy(mgr). + For(&networkingv1alpha.HTTPProxy{}). + Watches(&networkingv1alpha.NetworkService{}, enqueueEdgeReachabilityForNamespace()). + Watches(&networkingv1alpha.NetworkInterface{}, enqueueEdgeReachabilityForNamespace()). + Named("edgereachability"). + Complete(r) +} + +// enqueueEdgeReachabilityForNamespace enqueues the one record a namespace +// holds. Reconcile reads the namespace off the request and recomputes the whole +// answer, so the name carried here only has to be stable. +func enqueueEdgeReachabilityForNamespace() func( + clusterName multicluster.ClusterName, + cl cluster.Cluster, +) handler.TypedEventHandler[client.Object, mcreconcile.Request] { + return func(clusterName multicluster.ClusterName, _ cluster.Cluster) handler.TypedEventHandler[client.Object, mcreconcile.Request] { + return handler.TypedEnqueueRequestsFromMapFunc(func(_ context.Context, obj client.Object) []mcreconcile.Request { + if obj.GetNamespace() == "" { + return nil + } + return []mcreconcile.Request{{ + ClusterName: clusterName, + Request: ctrl.Request{NamespacedName: client.ObjectKey{ + Namespace: obj.GetNamespace(), + Name: networkingv1alpha.EdgeReachabilityName, + }}, + }} + }) + } +} diff --git a/internal/controller/edgereachability_controller_test.go b/internal/controller/edgereachability_controller_test.go new file mode 100644 index 00000000..02e59a5c --- /dev/null +++ b/internal/controller/edgereachability_controller_test.go @@ -0,0 +1,292 @@ +// SPDX-License-Identifier: AGPL-3.0-only + +package controller + +import ( + "context" + "strings" + "testing" + + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + discoveryv1 "k8s.io/api/discovery/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" +) + +// exposure stands a project control plane and the federation hub on two API +// servers, which is what they are: the control plane holds the proxies and the +// interfaces, and the hub is the only plane a cell can also read. +type exposure struct { + t *testing.T + ctx context.Context + + project client.Client + hub client.Client + + projectNamespace string + hubNamespace string + + reconciler *EdgeReachabilityReconciler +} + +func newExposure(t *testing.T) *exposure { + t.Helper() + + projectPlane, hubPlane := startPlanes(t) + ctx := context.Background() + + namespace := &corev1.Namespace{} + namespace.Name = "proj-" + sanitizeName(strings.ToLower(t.Name())) + require.NoError(t, projectPlane.Create(ctx, namespace)) + + hubNamespace := &corev1.Namespace{} + hubNamespace.Name = "ns-" + string(namespace.UID) + require.NoError(t, hubPlane.Create(ctx, hubNamespace)) + + return &exposure{ + t: t, + ctx: ctx, + project: projectPlane, + hub: hubPlane, + projectNamespace: namespace.Name, + hubNamespace: hubNamespace.Name, + reconciler: &EdgeReachabilityReconciler{ + DownstreamCluster: &hubFakeCluster{scheme: hubPlane.Scheme(), client: hubPlane}, + }, + } +} + +func (e *exposure) record() { + e.t.Helper() + require.NoError(e.t, e.reconciler.record(e.ctx, "cluster-"+testProject, e.project, e.projectNamespace)) +} + +func (e *exposure) recorded() (*networkingv1alpha.EdgeReachability, bool) { + e.t.Helper() + + var record networkingv1alpha.EdgeReachability + err := e.hub.Get(e.ctx, client.ObjectKey{ + Namespace: e.hubNamespace, + Name: networkingv1alpha.EdgeReachabilityName, + }, &record) + if err != nil { + return nil, false + } + return &record, true +} + +func (e *exposure) interfaceHolding(name, address string, labels map[string]string) { + e.t.Helper() + + iface := &networkingv1alpha.NetworkInterface{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: e.projectNamespace, + Name: name, + Labels: labels, + }, + Spec: networkingv1alpha.NetworkInterfaceSpec{ + Network: networkingv1alpha.LocalNetworkRef{Name: "default"}, + Addresses: []networkingv1alpha.NetworkInterfaceAddress{{ + Address: address + "/128", + Family: networkingv1alpha.IPv6Protocol, + Primary: true, + }}, + }, + } + require.NoError(e.t, e.project.Create(e.ctx, iface)) + + iface.Status.Phase = networkingv1alpha.NetworkInterfacePhaseBound + require.NoError(e.t, e.project.Status().Update(e.ctx, iface)) +} + +func (e *exposure) service(name string, matchLabels map[string]string) { + e.t.Helper() + + service := &networkingv1alpha.NetworkService{ + ObjectMeta: metav1.ObjectMeta{Namespace: e.projectNamespace, Name: name}, + Spec: networkingv1alpha.NetworkServiceSpec{ + NetworkInterfaces: networkingv1alpha.NetworkServiceInterfaceSelector{ + Selector: metav1.LabelSelector{MatchLabels: matchLabels}, + }, + Ports: []networkingv1alpha.NetworkServicePort{{Name: "http", Port: 8080}}, + }, + } + require.NoError(e.t, e.project.Create(e.ctx, service)) +} + +func (e *exposure) proxyBackedByWeb() { + e.t.Helper() + + proxy := &networkingv1alpha.HTTPProxy{ + ObjectMeta: metav1.ObjectMeta{Namespace: e.projectNamespace, Name: "site"}, + Spec: networkingv1alpha.HTTPProxySpec{ + Rules: []networkingv1alpha.HTTPProxyRule{{ + Backends: []networkingv1alpha.HTTPProxyRuleBackend{{ + NetworkService: &networkingv1alpha.NetworkServiceBackendRef{ + Name: "web", + Port: "http", + }, + }}, + }}, + }, + } + require.NoError(e.t, e.project.Create(e.ctx, proxy)) +} + +// A project with workloads and no proxy is the ordinary case, and it is the one +// that put every tenant's pods on every edge. The record has to say so rather +// than say nothing. +func TestAProjectWithNoProxyRecordsAnEmptyAnswer(t *testing.T) { + e := newExposure(t) + e.interfaceHolding("web-0", "fd20:0:2::1:0:0", map[string]string{"compute.datumapis.com/workload-name": "web"}) + + e.record() + + record, found := e.recorded() + require.True(t, found, "an answer of none is still an answer") + require.Empty(t, record.Spec.Addresses) +} + +func TestOnlyTheMembersOfAProxiedServiceAreRecorded(t *testing.T) { + e := newExposure(t) + e.interfaceHolding("web-0", "fd20:0:2::1:0:0", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.interfaceHolding("batch-0", "fd20:0:2::2:0:0", map[string]string{"compute.datumapis.com/workload-name": "batch"}) + e.service("web", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.service("batch", map[string]string{"compute.datumapis.com/workload-name": "batch"}) + e.proxyBackedByWeb() + + e.record() + + record, found := e.recorded() + require.True(t, found) + require.Equal(t, []string{"fd20:0:2::1:0:0"}, record.Spec.Addresses, + "a service nothing proxies puts nothing on an edge") +} + +// The address is what a cell joins on, so a prefix length carried into the +// record would match nothing and withdraw a pod that is serving. +func TestARecordedAddressCarriesNoPrefixLength(t *testing.T) { + e := newExposure(t) + e.interfaceHolding("web-0", "fd20:0:2::1:0:0", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.service("web", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.proxyBackedByWeb() + + e.record() + + record, _ := e.recorded() + require.Equal(t, []string{"fd20:0:2::1:0:0"}, record.Spec.Addresses) +} + +// A selector narrowed, or a workload scaled in, has to take the address back +// out. A record that only ever grew would keep every pod a project had ever run +// reachable from every edge. +func TestARecordFollowsASelectorThatStopsMatching(t *testing.T) { + e := newExposure(t) + e.interfaceHolding("web-0", "fd20:0:2::1:0:0", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.service("web", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.proxyBackedByWeb() + + e.record() + record, _ := e.recorded() + require.Len(t, record.Spec.Addresses, 1) + + var iface networkingv1alpha.NetworkInterface + require.NoError(t, e.project.Get(e.ctx, client.ObjectKey{ + Namespace: e.projectNamespace, Name: "web-0", + }, &iface)) + iface.Labels["compute.datumapis.com/workload-name"] = "retired" + require.NoError(t, e.project.Update(e.ctx, &iface)) + + e.record() + + record, _ = e.recorded() + require.Empty(t, record.Spec.Addresses) +} + +// An interface no workload holds is retired capacity. Nothing answers on its +// addresses, so no edge needs a route to them. +func TestAnUnheldInterfaceIsNotRecorded(t *testing.T) { + e := newExposure(t) + e.service("web", map[string]string{"compute.datumapis.com/workload-name": "web"}) + e.proxyBackedByWeb() + + iface := &networkingv1alpha.NetworkInterface{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: e.projectNamespace, + Name: "web-0", + Labels: map[string]string{"compute.datumapis.com/workload-name": "web"}, + }, + Spec: networkingv1alpha.NetworkInterfaceSpec{ + Network: networkingv1alpha.LocalNetworkRef{Name: "default"}, + Addresses: []networkingv1alpha.NetworkInterfaceAddress{{ + Address: "fd20:0:2::1:0:0/128", + Family: networkingv1alpha.IPv6Protocol, + Primary: true, + }}, + }, + } + require.NoError(t, e.project.Create(e.ctx, iface)) + + e.record() + + record, _ := e.recorded() + require.Empty(t, record.Spec.Addresses) +} + +// A proxy naming a pod's slice directly reaches that pod without a service, and +// dropping it out of the record would black-hole a backend that works today. +func TestAnInstanceBackendIsRecorded(t *testing.T) { + e := newExposure(t) + + slice := &discoveryv1.EndpointSlice{ + ObjectMeta: metav1.ObjectMeta{Namespace: e.projectNamespace, Name: "pod-0"}, + AddressType: discoveryv1.AddressTypeIPv6, + Endpoints: []discoveryv1.Endpoint{{ + Addresses: []string{"fd20:0:2::7:0:0"}, + }}, + } + require.NoError(t, e.project.Create(e.ctx, slice)) + + proxy := &networkingv1alpha.HTTPProxy{ + ObjectMeta: metav1.ObjectMeta{Namespace: e.projectNamespace, Name: "site"}, + Spec: networkingv1alpha.HTTPProxySpec{ + Rules: []networkingv1alpha.HTTPProxyRule{{ + Backends: []networkingv1alpha.HTTPProxyRuleBackend{{ + Instance: &networkingv1alpha.InstanceBackendRef{ + Name: "pod-0", + Port: 8080, + }, + }}, + }}, + }, + } + require.NoError(t, e.project.Create(e.ctx, proxy)) + + e.record() + + record, _ := e.recorded() + require.Equal(t, []string{"fd20:0:2::7:0:0"}, record.Spec.Addresses) +} + +// The hub namespace is made by the federation that carries a project's work +// out. A namespace that is not there yet holds nothing to withdraw, and +// inventing one would leave an object nothing collects. +func TestNoRecordIsWrittenWithoutAHubNamespace(t *testing.T) { + e := newExposure(t) + + orphan := &corev1.Namespace{} + orphan.Name = "proj-unfederated-" + sanitizeName(strings.ToLower(t.Name())) + require.NoError(t, e.project.Create(e.ctx, orphan)) + + require.NoError(t, e.reconciler.record(e.ctx, "cluster-"+testProject, e.project, orphan.Name)) + + var record networkingv1alpha.EdgeReachability + err := e.hub.Get(e.ctx, client.ObjectKey{ + Namespace: "ns-" + string(orphan.UID), + Name: networkingv1alpha.EdgeReachabilityName, + }, &record) + require.Error(t, err) +} diff --git a/internal/controller/networkinterfaceclaim_controller.go b/internal/controller/networkinterfaceclaim_controller.go index 487634fd..b8f81348 100644 --- a/internal/controller/networkinterfaceclaim_controller.go +++ b/internal/controller/networkinterfaceclaim_controller.go @@ -262,7 +262,7 @@ func (r *NetworkInterfaceClaimReconciler) resolveProject( func resolveProjectRouting( ctx context.Context, - cl client.Client, + cl client.Reader, namespaceName string, ) (projectRouting, error) { var namespace corev1.Namespace @@ -294,7 +294,7 @@ func resolveProjectRouting( // names. func resolveProjectOrCluster( ctx context.Context, - cl client.Client, + cl client.Reader, namespaceName string, ) (projectRouting, error) { routing, err := resolveProjectRouting(ctx, cl, namespaceName) diff --git a/internal/controller/vpcendpointslice_writeback.go b/internal/controller/vpcendpointslice_writeback.go index 439dba06..f5a87d69 100644 --- a/internal/controller/vpcendpointslice_writeback.go +++ b/internal/controller/vpcendpointslice_writeback.go @@ -54,6 +54,12 @@ const ( // went while the cell could not see the hub. Nothing replays a deletion. const vpcEndpointSliceSweepInterval = 10 * time.Minute +// vpcEndpointSliceResyncInterval paces the pass that acts on a change only the +// hub saw. A pod put behind a proxy, or taken out from behind one, moves a +// record on the hub and nothing in the cell. Without a pass of its own, a pod +// would wait for an unrelated local event before the edge could reach it. +const vpcEndpointSliceResyncInterval = time.Minute + // VPCEndpointSliceWriteBackReconciler publishes the per-pod EndpointSlices // galactic-cni writes in a cell to the federation hub, so the propagation // policy already selecting EndpointSlices carries them to every gateway @@ -76,6 +82,7 @@ type VPCEndpointSliceWriteBackReconciler struct { } // +kubebuilder:rbac:groups=discovery.k8s.io,resources=endpointslices,verbs=get;list;watch;create;update;patch;delete +// +kubebuilder:rbac:groups=networking.datumapis.com,resources=edgereachabilities,verbs=get;list;watch func (r *VPCEndpointSliceWriteBackReconciler) Reconcile(ctx context.Context, req mcreconcile.Request) (ctrl.Result, error) { cl, err := r.mgr.GetCluster(ctx, req.ClusterName) @@ -88,7 +95,7 @@ func (r *VPCEndpointSliceWriteBackReconciler) Reconcile(ctx context.Context, req func (r *VPCEndpointSliceWriteBackReconciler) publish( ctx context.Context, - cl client.Client, + cl client.Reader, key client.ObjectKey, ) error { logger := log.FromContext(ctx) @@ -135,6 +142,21 @@ func (r *VPCEndpointSliceWriteBackReconciler) publish( return nil } + // A pod nothing serves through a proxy is a pod no edge has to reach, and + // carrying it puts one tenant's addresses on every other tenant's edge. + // + // Silence is not a withdrawal. A namespace the control plane has not + // answered for keeps whatever it already reaches: a route pulled out from + // under a pod that is still serving black-holes live traffic, while a route + // left up for a pod nothing sends to costs a table entry. + reachable, recorded, err := r.reachability(ctx, hub, slice.Namespace) + if err != nil { + return err + } + if recorded && !servesReachableAddress(&slice, reachable) { + return r.collect(ctx, hub, location, key) + } + // A copy exists to be federated onward. A slice whose namespace names no // project cannot be routed and would leave an object nothing collects. routing, err := resolveProjectRouting(ctx, cl, slice.Namespace) @@ -167,6 +189,43 @@ func (r *VPCEndpointSliceWriteBackReconciler) publish( return nil } +// reachability reads the control plane's answer for a namespace: which of the +// project's workload addresses are behind a proxy. The second return says +// whether an answer exists at all, which no set of addresses can express. +func (r *VPCEndpointSliceWriteBackReconciler) reachability( + ctx context.Context, + hub client.Reader, + namespace string, +) (map[string]struct{}, bool, error) { + var record networkingv1alpha.EdgeReachability + key := client.ObjectKey{Namespace: namespace, Name: networkingv1alpha.EdgeReachabilityName} + if err := hub.Get(ctx, key, &record); err != nil { + if apierrors.IsNotFound(err) { + return nil, false, nil + } + return nil, false, fmt.Errorf("failed reading edge reachability for %q: %w", namespace, err) + } + + addresses := make(map[string]struct{}, len(record.Spec.Addresses)) + for _, address := range record.Spec.Addresses { + addresses[address] = struct{}{} + } + return addresses, true, nil +} + +// servesReachableAddress reports whether the slice describes an address the +// control plane says something serves. +func servesReachableAddress(slice *discoveryv1.EndpointSlice, reachable map[string]struct{}) bool { + for _, endpoint := range slice.Endpoints { + for _, address := range endpoint.Addresses { + if _, ok := reachable[address]; ok { + return true + } + } + } + return false +} + // isVPCPodEndpointSlice reports whether a slice is one galactic-cni published // for a VPC pod. func isVPCPodEndpointSlice(slice *discoveryv1.EndpointSlice) bool { @@ -369,6 +428,30 @@ func (r *VPCEndpointSliceWriteBackReconciler) sweep(ctx context.Context) error { return errors.Join(errs...) } +// resync republishes every vpc slice the cell holds, so a reachability record +// that has changed on the hub is acted on without waiting for the pod to change. +// publish decides both directions, so this both restores a copy that was +// withheld and removes one that is no longer served. +func (r *VPCEndpointSliceWriteBackReconciler) resync(ctx context.Context) error { + var held discoveryv1.EndpointSliceList + if err := r.localReader.List(ctx, &held, client.HasLabels{VPCPodTenantIDLabel}); err != nil { + return fmt.Errorf("failed listing vpc endpointslices: %w", err) + } + + var errs []error + for i := range held.Items { + slice := &held.Items[i] + if isVPCEndpointSliceCopy(slice) { + continue + } + if err := r.publish(ctx, r.localReader, client.ObjectKeyFromObject(slice)); err != nil { + errs = append(errs, err) + } + } + + return errors.Join(errs...) +} + func (r *VPCEndpointSliceWriteBackReconciler) location(ctx context.Context) (string, error) { identity, err := ResolveLocationIdentity(ctx, r.localReader, r.Location) if err != nil { @@ -383,17 +466,24 @@ func (r *VPCEndpointSliceWriteBackReconciler) location(ctx context.Context) (str // Start runs the sweep on a timer for as long as the manager runs. func (r *VPCEndpointSliceWriteBackReconciler) Start(ctx context.Context) error { - ticker := time.NewTicker(vpcEndpointSliceSweepInterval) - defer ticker.Stop() + sweeps := time.NewTicker(vpcEndpointSliceSweepInterval) + defer sweeps.Stop() + + resyncs := time.NewTicker(vpcEndpointSliceResyncInterval) + defer resyncs.Stop() for { select { case <-ctx.Done(): return nil - case <-ticker.C: + case <-sweeps.C: if err := r.sweep(ctx); err != nil { log.FromContext(ctx).Error(err, "failed sweeping published vpc endpointslices") } + case <-resyncs.C: + if err := r.resync(ctx); err != nil { + log.FromContext(ctx).Error(err, "failed resyncing published vpc endpointslices") + } } } } diff --git a/internal/controller/vpcendpointslice_writeback_test.go b/internal/controller/vpcendpointslice_writeback_test.go index 4d045b04..f4cbfe04 100644 --- a/internal/controller/vpcendpointslice_writeback_test.go +++ b/internal/controller/vpcendpointslice_writeback_test.go @@ -396,3 +396,111 @@ func TestALongPodNameStillPublishes(t *testing.T) { require.NoError(t, r.writeBack.sweep(r.ctx)) require.NoError(t, r.hub.Get(r.ctx, client.ObjectKeyFromObject(&copied), &copied)) } + +// recordEdgeReachability writes the control plane's answer for this namespace +// onto the hub, which is where a cell reads it from. +func (r *reachability) recordEdgeReachability(addresses ...string) { + r.t.Helper() + + record := &networkingv1alpha.EdgeReachability{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: r.namespace, + Name: networkingv1alpha.EdgeReachabilityName, + }, + Spec: networkingv1alpha.EdgeReachabilitySpec{Addresses: addresses}, + } + require.NoError(r.t, r.hub.Create(r.ctx, record)) +} + +// A namespace the control plane has not answered for keeps reaching what it +// already reached. Treating silence as a withdrawal would pull the route out +// from under every pod in a project the moment the record stopped arriving. +func TestASliceIsPublishedWhileTheControlPlaneHasNotAnswered(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + + r.publish(liveSliceName) + + _, found := r.hubCopy() + require.True(t, found, "no answer is not an answer of no") +} + +func TestASliceServingAProxyIsPublished(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.recordEdgeReachability("fd20:0:2::9:0:0", liveAddress) + + r.publish(liveSliceName) + + _, found := r.hubCopy() + require.True(t, found) +} + +// The whole point: a pod nothing serves through a proxy is a pod no edge has +// to reach, and carrying it puts one tenant's addresses on every other +// tenant's edge. +func TestASliceServingNoProxyIsNotPublished(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.recordEdgeReachability("fd20:0:2::9:0:0") + + r.publish(liveSliceName) + + _, found := r.hubCopy() + require.False(t, found, "a pod behind no proxy is not carried to the edge") +} + +// A project that publishes nothing is a real answer, and it is the answer most +// projects give. +func TestAnEmptyAnswerWithdrawsEverything(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.recordEdgeReachability() + + r.publish(liveSliceName) + + _, found := r.hubCopy() + require.False(t, found) +} + +// A proxy deleted, or a selector narrowed, leaves a copy behind that nothing +// routes to. The next pass has to take it back down. +func TestACopyIsCollectedWhenItStopsServingAProxy(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + + r.publish(liveSliceName) + _, found := r.hubCopy() + require.True(t, found) + + r.recordEdgeReachability() + r.publish(liveSliceName) + + _, found = r.hubCopy() + require.False(t, found, "a copy nothing serves is collected") +} + +// Nothing in a cell watches the hub, so a pod put behind a proxy has no local +// event to act on. The resync is the only thing that brings it back. +func TestTheResyncPublishesASliceThatHasJustBeenPutBehindAProxy(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.recordEdgeReachability() + + r.publish(liveSliceName) + _, found := r.hubCopy() + require.False(t, found) + + var record networkingv1alpha.EdgeReachability + require.NoError(t, r.hub.Get(r.ctx, client.ObjectKey{ + Namespace: r.namespace, + Name: networkingv1alpha.EdgeReachabilityName, + }, &record)) + record.Spec.Addresses = []string{liveAddress} + require.NoError(t, r.hub.Update(r.ctx, &record)) + + require.NoError(t, r.writeBack.resync(r.ctx)) + + _, found = r.hubCopy() + require.True(t, found, "the resync is what acts on a change only the hub saw") +}