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") +}