From 57599bb888574cfbc136a8b5ad0e05de66bd21bb Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Tue, 10 Nov 2026 14:41:08 +0000 Subject: [PATCH 1/5] fix: publish vpc records on reachability change The cell write-back learned that a pod had been put behind a proxy only on a one-minute resync, so a new cross-location proxy waited up to a minute before the edge could reach its backend. It now watches the hub's edge reachability records and republishes that namespace's slices the moment one changes. Key changes: - watch EdgeReachability on the hub and republish only its namespace - drop the one-minute resync of every slice - run the sweep for orphaned copies once at startup, with backoff, instead of every ten minutes; a deletion seen while the hub is down is retried by the workqueue, so only a restart can miss one --- .../controller/vpcendpointslice_writeback.go | 90 ++++++++++++------- .../vpcendpointslice_writeback_test.go | 18 ++-- 2 files changed, 71 insertions(+), 37 deletions(-) diff --git a/internal/controller/vpcendpointslice_writeback.go b/internal/controller/vpcendpointslice_writeback.go index fa79e39f..047fa7e6 100644 --- a/internal/controller/vpcendpointslice_writeback.go +++ b/internal/controller/vpcendpointslice_writeback.go @@ -18,8 +18,11 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/cluster" "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + "sigs.k8s.io/controller-runtime/pkg/source" mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder" mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile" @@ -50,20 +53,14 @@ const ( karmadaManagedLabel = "karmada.io/managed" ) -// vpcEndpointSliceSweepInterval paces the collection of copies whose source -// went while the cell could not see the hub. Nothing replays a deletion. -const vpcEndpointSliceSweepInterval = 10 * time.Minute +// vpcEndpointSliceSweepRetryLimit caps the wait between attempts at the +// startup sweep while the hub or the cell cannot yet be read. +const vpcEndpointSliceSweepRetryLimit = time.Minute // vpcEndpointSliceFieldManager owns the fields the write-back sets on a copy, // and no others. const vpcEndpointSliceFieldManager = "network-services-operator/vpc-endpointslice-writeback" -// 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 @@ -433,14 +430,19 @@ 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 { +// republishNamespace republishes every vpc slice the cell holds in a +// namespace, so a reachability record that changed on the hub is acted on +// without waiting for the pod to change. A pod put behind a proxy, or taken out +// from behind one, moves the record and nothing in the cell. publish decides +// both directions, so this both restores a copy that was withheld and removes +// one that is no longer served. +func (r *VPCEndpointSliceWriteBackReconciler) republishNamespace(ctx context.Context, namespace string) 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) + if err := r.localReader.List(ctx, &held, + client.InNamespace(namespace), + client.HasLabels{VPCPodTenantIDLabel}, + ); err != nil { + return fmt.Errorf("failed listing vpc endpointslices in %q: %w", namespace, err) } var errs []error @@ -469,30 +471,45 @@ func (r *VPCEndpointSliceWriteBackReconciler) location(ctx context.Context) (str return identity.Reference.Name, nil } -// Start runs the sweep on a timer for as long as the manager runs. +// Start sweeps once, when the process starts, retrying until it succeeds. +// +// A deletion is only missed while nothing is watching: a deletion seen while +// the hub is unreachable stays queued and is retried until it lands. So the +// only gap is a restart, and that is where the sweep runs. func (r *VPCEndpointSliceWriteBackReconciler) Start(ctx context.Context) error { - sweeps := time.NewTicker(vpcEndpointSliceSweepInterval) - defer sweeps.Stop() - - resyncs := time.NewTicker(vpcEndpointSliceResyncInterval) - defer resyncs.Stop() - + wait := time.Second for { + err := r.sweep(ctx) + if err == nil { + return nil + } + log.FromContext(ctx).Error(err, "failed sweeping published vpc endpointslices", "retryIn", wait.String()) + select { case <-ctx.Done(): return nil - 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") - } + case <-time.After(wait): } + wait = min(2*wait, vpcEndpointSliceSweepRetryLimit) } } +// edgeReachabilityNamespace maps a reachability record on the hub to the +// namespace whose slices it decides. +func edgeReachabilityNamespace(_ context.Context, record *networkingv1alpha.EdgeReachability) []reconcile.Request { + return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: record.Namespace}}} +} + +// vpcEndpointSliceReachabilityReconciler republishes a namespace's slices when +// the hub's reachability record for it changes. +type vpcEndpointSliceReachabilityReconciler struct { + writeBack *VPCEndpointSliceWriteBackReconciler +} + +func (r *vpcEndpointSliceReachabilityReconciler) Reconcile(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { + return ctrl.Result{}, r.writeBack.republishNamespace(ctx, req.Name) +} + // vpcPodEndpointSlicePredicate passes the slices this write-back carries. An // update is passed when either side carries the tenant label, so that a slice // which loses it still reaches Reconcile and has its copy collected. @@ -525,6 +542,17 @@ func (r *VPCEndpointSliceWriteBackReconciler) SetupWithManager(mgr mcmanager.Man return fmt.Errorf("unable to add the vpc endpointslice sweep: %w", err) } + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). + WatchesRawSource(source.Kind( + r.HubCluster.GetCache(), + &networkingv1alpha.EdgeReachability{}, + handler.TypedEnqueueRequestsFromMapFunc(edgeReachabilityNamespace), + )). + Named("vpcendpointslice_writeback_reachability"). + Complete(&vpcEndpointSliceReachabilityReconciler{writeBack: r}); err != nil { + return fmt.Errorf("unable to watch edge reachability on the hub: %w", err) + } + return mcbuilder.ControllerManagedBy(mgr). For( &discoveryv1.EndpointSlice{}, diff --git a/internal/controller/vpcendpointslice_writeback_test.go b/internal/controller/vpcendpointslice_writeback_test.go index ced338ad..cf4a4a26 100644 --- a/internal/controller/vpcendpointslice_writeback_test.go +++ b/internal/controller/vpcendpointslice_writeback_test.go @@ -13,6 +13,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/utils/ptr" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" "go.datum.net/network-services-operator/internal/config" @@ -481,9 +482,9 @@ func TestACopyIsCollectedWhenItStopsServingAProxy(t *testing.T) { 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) { +// A pod put behind a proxy changes the hub's reachability record and nothing in +// the cell. The record's own change is what brings the slice back. +func TestAReachabilityChangePublishesASliceThatHasJustBeenPutBehindAProxy(t *testing.T) { r := newReachability(t) r.sliceOnCell() r.recordEdgeReachability() @@ -500,10 +501,15 @@ func TestTheResyncPublishesASliceThatHasJustBeenPutBehindAProxy(t *testing.T) { record.Spec.Addresses = []string{liveAddress} require.NoError(t, r.hub.Update(r.ctx, &record)) - require.NoError(t, r.writeBack.resync(r.ctx)) + requests := edgeReachabilityNamespace(r.ctx, &record) + require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{Name: r.namespace}}}, requests) + + reachability := &vpcEndpointSliceReachabilityReconciler{writeBack: r.writeBack} + _, err := reachability.Reconcile(r.ctx, requests[0]) + require.NoError(t, err) _, found = r.hubCopy() - require.True(t, found, "the resync is what acts on a change only the hub saw") + require.True(t, found, "the record's change is what acts on a change only the hub saw") } // hubClaimLabel is the label Karmada writes onto every object its propagation @@ -512,7 +518,7 @@ const hubClaimLabel = "clusterpropagationpolicy.karmada.io/permanent-id" // The hub claims a copy by writing its own label onto it. A publish must leave // that label alone: removing it makes the hub write it back, so the copy would -// change twice on every resync while its source does not change at all. +// change twice on every republish while its source does not change at all. func TestTheHubsClaimOnACopySurvivesThePublish(t *testing.T) { r := newReachability(t) r.sliceOnCell() From fc8ae10aae79bb2c76f8dfbfa41006850332751d Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Tue, 10 Nov 2026 14:41:08 +0000 Subject: [PATCH 2/5] test: cover reachability-driven publishing end to end The one-minute resync also restored a hub copy deleted or edited out from under the cell. Watch the cell's own copies on the hub so that is now an event too, and add an e2e scenario against the real cell and hub that fails if any of these waits for a periodic pass. Key changes: - watch this location's copies on the hub and republish their source - e2e: a slice is published, restored and withdrawn within 10s of the reachability record or the hub copy changing --- .../controller/vpcendpointslice_writeback.go | 48 +++++- .../vpcendpointslice_writeback_test.go | 41 ++++- .../chainsaw-test.yaml | 156 ++++++++++++++++++ 3 files changed, 236 insertions(+), 9 deletions(-) create mode 100644 test/e2e/vpcendpointslice-reachability-event/chainsaw-test.yaml diff --git a/internal/controller/vpcendpointslice_writeback.go b/internal/controller/vpcendpointslice_writeback.go index 047fa7e6..f2b418f0 100644 --- a/internal/controller/vpcendpointslice_writeback.go +++ b/internal/controller/vpcendpointslice_writeback.go @@ -500,14 +500,37 @@ func edgeReachabilityNamespace(_ context.Context, record *networkingv1alpha.Edge return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: record.Namespace}}} } -// vpcEndpointSliceReachabilityReconciler republishes a namespace's slices when -// the hub's reachability record for it changes. -type vpcEndpointSliceReachabilityReconciler struct { - writeBack *VPCEndpointSliceWriteBackReconciler +// reconcileReachability republishes a namespace's slices when the hub's +// reachability record for it changes. +func (r *VPCEndpointSliceWriteBackReconciler) reconcileReachability(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { + return ctrl.Result{}, r.republishNamespace(ctx, req.Name) } -func (r *vpcEndpointSliceReachabilityReconciler) Reconcile(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { - return ctrl.Result{}, r.writeBack.republishNamespace(ctx, req.Name) +// copySource maps a copy this cell published to the slice it came from, so a +// copy deleted or edited on the hub is put back as the cell says it should be. +func (r *VPCEndpointSliceWriteBackReconciler) copySource(ctx context.Context, copied *discoveryv1.EndpointSlice) []reconcile.Request { + if copied.Labels[VPCEndpointSliceProjectionLabel] != "true" { + return nil + } + source := copied.Annotations[VPCEndpointSliceSourceNameAnnotation] + if source == "" { + return nil + } + location, err := r.location(ctx) + if err != nil { + log.FromContext(ctx).Error(err, "failed resolving the location of a changed vpc endpointslice copy", + "namespace", copied.Namespace, jsonKeyName, copied.Name) + return nil + } + if location == "" || copied.Labels[networkingv1alpha.NetworkInterfaceLocationLabel] != location { + return nil + } + return []reconcile.Request{{NamespacedName: client.ObjectKey{Namespace: copied.Namespace, Name: source}}} +} + +// reconcileCopy republishes the slice behind a copy that changed on the hub. +func (r *VPCEndpointSliceWriteBackReconciler) reconcileCopy(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { + return ctrl.Result{}, r.publish(ctx, r.localReader, req.NamespacedName) } // vpcPodEndpointSlicePredicate passes the slices this write-back carries. An @@ -549,10 +572,21 @@ func (r *VPCEndpointSliceWriteBackReconciler) SetupWithManager(mgr mcmanager.Man handler.TypedEnqueueRequestsFromMapFunc(edgeReachabilityNamespace), )). Named("vpcendpointslice_writeback_reachability"). - Complete(&vpcEndpointSliceReachabilityReconciler{writeBack: r}); err != nil { + Complete(reconcile.Func(r.reconcileReachability)); err != nil { return fmt.Errorf("unable to watch edge reachability on the hub: %w", err) } + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). + WatchesRawSource(source.Kind( + r.HubCluster.GetCache(), + &discoveryv1.EndpointSlice{}, + handler.TypedEnqueueRequestsFromMapFunc(r.copySource), + )). + Named("vpcendpointslice_writeback_copies"). + Complete(reconcile.Func(r.reconcileCopy)); err != nil { + return fmt.Errorf("unable to watch published vpc endpointslices on the hub: %w", err) + } + return mcbuilder.ControllerManagedBy(mgr). For( &discoveryv1.EndpointSlice{}, diff --git a/internal/controller/vpcendpointslice_writeback_test.go b/internal/controller/vpcendpointslice_writeback_test.go index cf4a4a26..bda7bef1 100644 --- a/internal/controller/vpcendpointslice_writeback_test.go +++ b/internal/controller/vpcendpointslice_writeback_test.go @@ -504,8 +504,7 @@ func TestAReachabilityChangePublishesASliceThatHasJustBeenPutBehindAProxy(t *tes requests := edgeReachabilityNamespace(r.ctx, &record) require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{Name: r.namespace}}}, requests) - reachability := &vpcEndpointSliceReachabilityReconciler{writeBack: r.writeBack} - _, err := reachability.Reconcile(r.ctx, requests[0]) + _, err := r.writeBack.reconcileReachability(r.ctx, requests[0]) require.NoError(t, err) _, found = r.hubCopy() @@ -557,3 +556,41 @@ func TestALabelTheSourceDropsLeavesTheCopy(t *testing.T) { copied, _ = r.hubCopy() require.NotContains(t, copied.Labels, "example.com/extra", "the dropped label leaves the copy") } + +// Nothing on the cell changes when a copy is deleted on the hub, so the copy's +// own deletion is what has to bring it back. +func TestACopyDeletedOnTheHubComesBack(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.publish(liveSliceName) + + copied, found := r.hubCopy() + require.True(t, found, "the cell's slice reaches the hub") + require.NoError(t, r.hub.Delete(r.ctx, copied)) + + requests := r.writeBack.copySource(r.ctx, copied) + require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{ + Namespace: r.namespace, + Name: liveSliceName, + }}}, requests) + + _, err := r.writeBack.reconcileCopy(r.ctx, requests[0]) + require.NoError(t, err) + + _, found = r.hubCopy() + require.True(t, found, "the deleted copy is published again") +} + +// Every cell publishes into the same hub namespaces, so a cell must only act on +// the copies it published itself. +func TestACopyFromAnotherLocationIsLeftToItsOwnCell(t *testing.T) { + r := newReachability(t) + r.sliceOnCell() + r.publish(liveSliceName) + + copied, found := r.hubCopy() + require.True(t, found, "the cell's slice reaches the hub") + copied.Labels[networkingv1alpha.NetworkInterfaceLocationLabel] = "elsewhere" + + require.Empty(t, r.writeBack.copySource(r.ctx, copied)) +} diff --git a/test/e2e/vpcendpointslice-reachability-event/chainsaw-test.yaml b/test/e2e/vpcendpointslice-reachability-event/chainsaw-test.yaml new file mode 100644 index 00000000..2062407c --- /dev/null +++ b/test/e2e/vpcendpointslice-reachability-event/chainsaw-test.yaml @@ -0,0 +1,156 @@ +# yaml-language-server: $schema=https://raw.githubusercontent.com/kyverno/chainsaw/main/.schemas/json/test-chainsaw-v1alpha1.json +# +# A cell carries a VPC pod's EndpointSlice to the hub only once the control +# plane's reachability record says a proxy sends traffic to it. Putting a pod +# behind a proxy changes that record on the hub and nothing on the cell, so the +# cell has to act on the record's own change. Every deadline here is well under +# a minute, so a cell that only noticed on a periodic pass fails. +# +# The scenario owns its namespace on both planes and writes the reachability +# record itself, so no control-plane reconcile can rewrite it mid-test. +apiVersion: chainsaw.kyverno.io/v1alpha1 +kind: Test +metadata: + name: vpcendpointslice-reachability-event +spec: + cluster: nso-standard + steps: + - name: A slice nothing serves stays on the cell + try: + - apply: + resource: + apiVersion: v1 + kind: Namespace + metadata: + name: vpc-reachability-e2e + labels: + meta.datumapis.com/upstream-cluster-name: cluster-project-alpha + meta.datumapis.com/upstream-namespace: default + - apply: + cluster: nso-hub + resource: + apiVersion: v1 + kind: Namespace + metadata: + name: vpc-reachability-e2e + - apply: + cluster: nso-hub + resource: + apiVersion: networking.datumapis.com/v1alpha + kind: EdgeReachability + metadata: + name: default + namespace: vpc-reachability-e2e + spec: + addresses: + - fd20:0:e2e::99 + - apply: + resource: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + metadata: + name: reachability-pod + namespace: vpc-reachability-e2e + labels: + galactic.datum.net/tenant-id: e2ereach-a1 + annotations: + galactic.datum.net/tenant-id: e2ereach-a1 + galactic.datum.net/srv6-sid: "2001:db8:e2e::" + addressType: IPv6 + endpoints: + - addresses: + - fd20:0:e2e::1 + conditions: + ready: true + - sleep: + duration: 10s + - error: + cluster: nso-hub + resource: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + metadata: + name: vpc-us-central-1-reachability-pod + namespace: vpc-reachability-e2e + catch: &diagnose + - script: + content: | + kubectl -n vpc-reachability-e2e get endpointslice -o yaml + kubectl -n network-services-operator-system logs -l app.kubernetes.io/component=cell-controller-manager --tail=100 + - script: + cluster: nso-hub + content: | + kubectl -n vpc-reachability-e2e get endpointslice,edgereachability -o yaml + + - name: Putting the pod behind a proxy publishes it within seconds + try: + - patch: + cluster: nso-hub + resource: + apiVersion: networking.datumapis.com/v1alpha + kind: EdgeReachability + metadata: + name: default + namespace: vpc-reachability-e2e + spec: + addresses: + - fd20:0:e2e::1 + - fd20:0:e2e::99 + - assert: + cluster: nso-hub + timeout: 10s + resource: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + metadata: + name: vpc-us-central-1-reachability-pod + namespace: vpc-reachability-e2e + labels: + networking.datumapis.com/location: us-central-1 + addressType: IPv6 + (endpoints[0].addresses[0]): fd20:0:e2e::1 + catch: *diagnose + + - name: A copy deleted on the hub comes back within seconds + try: + - delete: + cluster: nso-hub + ref: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + name: vpc-us-central-1-reachability-pod + namespace: vpc-reachability-e2e + - assert: + cluster: nso-hub + timeout: 10s + resource: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + metadata: + name: vpc-us-central-1-reachability-pod + namespace: vpc-reachability-e2e + catch: *diagnose + + - name: Taking the pod out from behind the proxy withdraws it within seconds + try: + - patch: + cluster: nso-hub + resource: + apiVersion: networking.datumapis.com/v1alpha + kind: EdgeReachability + metadata: + name: default + namespace: vpc-reachability-e2e + spec: + addresses: + - fd20:0:e2e::99 + - error: + cluster: nso-hub + timeout: 10s + resource: + apiVersion: discovery.k8s.io/v1 + kind: EndpointSlice + metadata: + name: vpc-us-central-1-reachability-pod + namespace: vpc-reachability-e2e + catch: *diagnose From 76f19687e0f7b33671d4f6e8064b848ecaa818d6 Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Tue, 10 Nov 2026 14:41:08 +0000 Subject: [PATCH 3/5] fix: wait on watches, not timers, for vpc records Every case where the write-back held a slice back used to be picked up again by the one-minute pass. Without it, each needs a watch on what it waits for. Key changes: - watch namespaces appearing on the hub, in the cell write-back and in the edge reachability writer, so a record held for a missing hub namespace is written once federation makes it - watch the cell's ServingLocation, when served, and republish and sweep once the location is known or changes - watch project labels on the cell's namespaces --- .../controller/edgereachability_controller.go | 41 +++++++- .../edgereachability_controller_test.go | 30 ++++++ .../controller/vpcendpointslice_writeback.go | 96 +++++++++++++++++-- .../vpcendpointslice_writeback_test.go | 34 ++++++- 4 files changed, 193 insertions(+), 8 deletions(-) diff --git a/internal/controller/edgereachability_controller.go b/internal/controller/edgereachability_controller.go index feaf119e..39b49a46 100644 --- a/internal/controller/edgereachability_controller.go +++ b/internal/controller/edgereachability_controller.go @@ -8,6 +8,7 @@ import ( "fmt" "slices" + corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -16,11 +17,14 @@ import ( "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/event" "sigs.k8s.io/controller-runtime/pkg/log" + "sigs.k8s.io/controller-runtime/pkg/predicate" 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" + mcsource "sigs.k8s.io/multicluster-runtime/pkg/source" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" "go.datum.net/network-services-operator/internal/downstreamclient" @@ -98,7 +102,7 @@ func (r *EdgeReachabilityReconciler) record( 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. + // there yet holds nothing to withdraw. Its creation records it. if apierrors.IsNotFound(err) { logger.Info("the hub has no namespace to record edge reachability in yet", "namespace", namespace) @@ -216,14 +220,49 @@ func (r *EdgeReachabilityReconciler) SetupWithManager(mgr mcmanager.Manager) err r.mgr = mgr + hubNamespaces, _, err := mcsource.TypedKind( + &corev1.Namespace{}, + enqueueEdgeReachabilityForHubNamespace, + predicate.TypedFuncs[*corev1.Namespace]{ + CreateFunc: func(event.TypedCreateEvent[*corev1.Namespace]) bool { return true }, + }, + ).ForCluster("", r.DownstreamCluster) + if err != nil { + return fmt.Errorf("failed to watch namespaces on the hub: %w", err) + } + return mcbuilder.ControllerManagedBy(mgr). For(&networkingv1alpha.HTTPProxy{}). + WatchesRawSource(hubNamespaces). Watches(&networkingv1alpha.NetworkService{}, enqueueEdgeReachabilityForNamespace()). Watches(&networkingv1alpha.NetworkInterface{}, enqueueEdgeReachabilityForNamespace()). Named("edgereachability"). Complete(r) } +// enqueueEdgeReachabilityForHubNamespace enqueues the record of the project +// namespace a hub namespace was made for, since a record can only be written +// once its hub namespace exists. +func enqueueEdgeReachabilityForHubNamespace( + _ multicluster.ClusterName, + _ cluster.Cluster, +) handler.TypedEventHandler[*corev1.Namespace, mcreconcile.Request] { + return handler.TypedEnqueueRequestsFromMapFunc(func(_ context.Context, namespace *corev1.Namespace) []mcreconcile.Request { + clusterLabel := namespace.Labels[downstreamclient.UpstreamOwnerClusterNameLabel] + upstreamNamespace := namespace.Labels[downstreamclient.UpstreamOwnerNamespaceLabel] + if clusterLabel == "" || upstreamNamespace == "" { + return nil + } + return []mcreconcile.Request{{ + ClusterName: multicluster.ClusterName(downstreamclient.UpstreamClusterNameFromLabel(clusterLabel)), + Request: ctrl.Request{NamespacedName: client.ObjectKey{ + Namespace: upstreamNamespace, + Name: networkingv1alpha.EdgeReachabilityName, + }}, + }} + }) +} + // 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. diff --git a/internal/controller/edgereachability_controller_test.go b/internal/controller/edgereachability_controller_test.go index e3b73b46..6819ec82 100644 --- a/internal/controller/edgereachability_controller_test.go +++ b/internal/controller/edgereachability_controller_test.go @@ -10,7 +10,11 @@ import ( corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile" networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" ) @@ -289,3 +293,29 @@ func TestNoRecordIsWrittenWithoutAHubNamespace(t *testing.T) { }, &record) require.Error(t, err) } + +// A record cannot be written before its hub namespace exists, so the namespace +// being made on the hub is what has to bring the record back. +func TestAHubNamespaceAppearingRecordsItsProject(t *testing.T) { + namespace := &corev1.Namespace{} + namespace.Name = "ns-0b6f" + namespace.Labels = map[string]string{ + "meta.datumapis.com/upstream-cluster-name": "cluster-_project-alpha", + "meta.datumapis.com/upstream-namespace": "default", + } + + queue := workqueue.NewTypedRateLimitingQueue(workqueue.DefaultTypedControllerRateLimiter[mcreconcile.Request]()) + defer queue.ShutDown() + enqueueEdgeReachabilityForHubNamespace("", nil).Create(context.Background(), + event.TypedCreateEvent[*corev1.Namespace]{Object: namespace}, queue) + + require.Equal(t, 1, queue.Len()) + got, _ := queue.Get() + require.Equal(t, mcreconcile.Request{ + ClusterName: "project-alpha", + Request: ctrl.Request{NamespacedName: client.ObjectKey{ + Namespace: "default", + Name: networkingv1alpha.EdgeReachabilityName, + }}, + }, got) +} diff --git a/internal/controller/vpcendpointslice_writeback.go b/internal/controller/vpcendpointslice_writeback.go index f2b418f0..360fb4f2 100644 --- a/internal/controller/vpcendpointslice_writeback.go +++ b/internal/controller/vpcendpointslice_writeback.go @@ -11,6 +11,7 @@ import ( "maps" "time" + corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -27,10 +28,12 @@ import ( mcmanager "sigs.k8s.io/multicluster-runtime/pkg/manager" mcreconcile "sigs.k8s.io/multicluster-runtime/pkg/reconcile" + locationsv1alpha1 "go.miloapis.com/locations/api/v1alpha1" + "go.miloapis.com/locations/pkg/locationidentity" + networkingv1alpha "go.datum.net/network-services-operator/api/v1alpha" "go.datum.net/network-services-operator/internal/config" "go.datum.net/network-services-operator/internal/downstreamclient" - "go.miloapis.com/locations/pkg/locationidentity" ) const ( @@ -178,7 +181,8 @@ func (r *VPCEndpointSliceWriteBackReconciler) publish( if err := writeFederatedEndpointSlice(ctx, hub, desired); err != nil { // The hub namespace is made by the federation that carries a - // project's work to this cell. Nothing here should invent one. + // project's work to this cell. Nothing here should invent one; its + // creation on the hub is what publishes this slice. if apierrors.IsNotFound(err) { logger.Info("the hub has no namespace to publish this vpc endpointslice into yet", "namespace", slice.Namespace) @@ -459,6 +463,9 @@ func (r *VPCEndpointSliceWriteBackReconciler) republishNamespace(ctx context.Con return errors.Join(errs...) } +// location is the location this cell publishes as, or empty while it has none. +// A ServingLocation being delivered is what publishes the slices held until +// then. func (r *VPCEndpointSliceWriteBackReconciler) location(ctx context.Context) (string, error) { identity, err := resolveLocationIdentity(ctx, r.localReader, r.Location) if err != nil { @@ -500,9 +507,43 @@ func edgeReachabilityNamespace(_ context.Context, record *networkingv1alpha.Edge return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: record.Namespace}}} } -// reconcileReachability republishes a namespace's slices when the hub's -// reachability record for it changes. -func (r *VPCEndpointSliceWriteBackReconciler) reconcileReachability(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { +// cellNamespace maps a cell namespace to itself. Its labels name the project a +// copy is routed to, so a namespace that gains them can publish what it holds. +func cellNamespace(_ context.Context, namespace *corev1.Namespace) []reconcile.Request { + return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: namespace.Name}}} +} + +// everyNamespace maps an object to one request that stands for every namespace. +func everyNamespace[T client.Object](context.Context, T) []reconcile.Request { + return []reconcile.Request{{}} +} + +// reconcileLocation republishes everything, and collects what has nothing +// behind it, once the location the cell publishes as is known or changes. +func (r *VPCEndpointSliceWriteBackReconciler) reconcileLocation(ctx context.Context, _ reconcile.Request) (ctrl.Result, error) { + return ctrl.Result{}, errors.Join(r.republishNamespace(ctx, ""), r.sweep(ctx)) +} + +// servesKind reports whether the cluster a manager reads serves a kind. A +// watch on a kind that is not served never syncs, and holds the manager with it. +func servesKind(mgr ctrl.Manager, obj client.Object) (bool, error) { + gvk, err := mgr.GetClient().GroupVersionKindFor(obj) + if err != nil { + return false, err + } + if _, err := mgr.GetRESTMapper().RESTMapping(gvk.GroupKind(), gvk.Version); err != nil { + if locationsNotServed(err) { + return false, nil + } + return false, fmt.Errorf("failed checking whether %s is served: %w", gvk.Kind, err) + } + return true, nil +} + +// reconcileNamespace republishes a namespace's slices when its reachability +// record or its namespace appears on the hub, or its project labels on the +// cell change. An empty name stands for every namespace. +func (r *VPCEndpointSliceWriteBackReconciler) reconcileNamespace(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { return ctrl.Result{}, r.republishNamespace(ctx, req.Name) } @@ -572,10 +613,53 @@ func (r *VPCEndpointSliceWriteBackReconciler) SetupWithManager(mgr mcmanager.Man handler.TypedEnqueueRequestsFromMapFunc(edgeReachabilityNamespace), )). Named("vpcendpointslice_writeback_reachability"). - Complete(reconcile.Func(r.reconcileReachability)); err != nil { + Complete(reconcile.Func(r.reconcileNamespace)); err != nil { return fmt.Errorf("unable to watch edge reachability on the hub: %w", err) } + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). + WatchesRawSource(source.Kind( + r.HubCluster.GetCache(), + &corev1.Namespace{}, + handler.TypedEnqueueRequestsFromMapFunc(cellNamespace), + predicate.TypedFuncs[*corev1.Namespace]{ + CreateFunc: func(event.TypedCreateEvent[*corev1.Namespace]) bool { return true }, + }, + )). + Named("vpcendpointslice_writeback_hub_namespaces"). + Complete(reconcile.Func(r.reconcileNamespace)); err != nil { + return fmt.Errorf("unable to watch namespaces on the hub: %w", err) + } + + servingLocations, err := servesKind(mgr.GetLocalManager(), &locationsv1alpha1.ServingLocation{}) + if err != nil { + return err + } + if servingLocations { + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). + WatchesRawSource(source.Kind( + mgr.GetLocalManager().GetCache(), + &locationsv1alpha1.ServingLocation{}, + handler.TypedEnqueueRequestsFromMapFunc(everyNamespace[*locationsv1alpha1.ServingLocation]), + )). + Named("vpcendpointslice_writeback_location"). + Complete(reconcile.Func(r.reconcileLocation)); err != nil { + return fmt.Errorf("unable to watch this cell's serving location: %w", err) + } + } + + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). + WatchesRawSource(source.Kind( + mgr.GetLocalManager().GetCache(), + &corev1.Namespace{}, + handler.TypedEnqueueRequestsFromMapFunc(cellNamespace), + predicate.TypedLabelChangedPredicate[*corev1.Namespace]{}, + )). + Named("vpcendpointslice_writeback_namespaces"). + Complete(reconcile.Func(r.reconcileNamespace)); err != nil { + return fmt.Errorf("unable to watch the cell's namespaces: %w", err) + } + if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). WatchesRawSource(source.Kind( r.HubCluster.GetCache(), diff --git a/internal/controller/vpcendpointslice_writeback_test.go b/internal/controller/vpcendpointslice_writeback_test.go index bda7bef1..700eb736 100644 --- a/internal/controller/vpcendpointslice_writeback_test.go +++ b/internal/controller/vpcendpointslice_writeback_test.go @@ -504,7 +504,7 @@ func TestAReachabilityChangePublishesASliceThatHasJustBeenPutBehindAProxy(t *tes requests := edgeReachabilityNamespace(r.ctx, &record) require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{Name: r.namespace}}}, requests) - _, err := r.writeBack.reconcileReachability(r.ctx, requests[0]) + _, err := r.writeBack.reconcileNamespace(r.ctx, requests[0]) require.NoError(t, err) _, found = r.hubCopy() @@ -594,3 +594,35 @@ func TestACopyFromAnotherLocationIsLeftToItsOwnCell(t *testing.T) { require.Empty(t, r.writeBack.copySource(r.ctx, copied)) } + +// A slice can reach the cell before federation has made its hub namespace. +// The namespace appearing on the hub is what publishes it. +func TestASliceIsPublishedOnceItsHubNamespaceAppears(t *testing.T) { + r := newReachability(t) + + ns := &corev1.Namespace{} + ns.Name = r.namespace + "-late" + ns.Labels = map[string]string{ + downstreamclient.UpstreamOwnerNamespaceLabel: "proj-" + scenarioSuffix(t), + downstreamclient.UpstreamOwnerClusterNameLabel: "cluster-" + testProject, + } + require.NoError(t, r.cell.Create(r.ctx, ns)) + r.namespace = ns.Name + r.sliceOnCell() + + r.publish(liveSliceName) + _, found := r.hubCopy() + require.False(t, found) + + hubNamespace := &corev1.Namespace{} + hubNamespace.Name = ns.Name + require.NoError(t, r.hub.Create(r.ctx, hubNamespace)) + + requests := cellNamespace(r.ctx, hubNamespace) + require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{Name: ns.Name}}}, requests) + _, err := r.writeBack.reconcileNamespace(r.ctx, requests[0]) + require.NoError(t, err) + + _, found = r.hubCopy() + require.True(t, found, "the hub namespace appearing publishes the slice") +} From 2fcd9b9eb2bfdfd3cf2617ce757d302c259185b6 Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Tue, 10 Nov 2026 14:41:08 +0000 Subject: [PATCH 4/5] refactor: drop the cell's hub namespace wait A cell's project namespace is propagated from its hub namespace, so a slice cannot reach the cell before the hub namespace exists. A missing one means the project is being torn down and there is nothing to wait for. Key changes: - remove the cell's watch on hub namespaces, which needed a hub permission the cell does not hold - remove the cell namespace label watch; labels arrive with the namespace --- .../controller/vpcendpointslice_writeback.go | 41 ++----------------- .../vpcendpointslice_writeback_test.go | 32 --------------- 2 files changed, 4 insertions(+), 69 deletions(-) diff --git a/internal/controller/vpcendpointslice_writeback.go b/internal/controller/vpcendpointslice_writeback.go index 360fb4f2..6c838948 100644 --- a/internal/controller/vpcendpointslice_writeback.go +++ b/internal/controller/vpcendpointslice_writeback.go @@ -11,7 +11,6 @@ import ( "maps" "time" - corev1 "k8s.io/api/core/v1" discoveryv1 "k8s.io/api/discovery/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -181,8 +180,9 @@ func (r *VPCEndpointSliceWriteBackReconciler) publish( if err := writeFederatedEndpointSlice(ctx, hub, desired); err != nil { // The hub namespace is made by the federation that carries a - // project's work to this cell. Nothing here should invent one; its - // creation on the hub is what publishes this slice. + // project's work to this cell, and the cell's own namespace is + // propagated from it, so a missing one is a project being torn down. + // Nothing here should invent one. if apierrors.IsNotFound(err) { logger.Info("the hub has no namespace to publish this vpc endpointslice into yet", "namespace", slice.Namespace) @@ -507,12 +507,6 @@ func edgeReachabilityNamespace(_ context.Context, record *networkingv1alpha.Edge return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: record.Namespace}}} } -// cellNamespace maps a cell namespace to itself. Its labels name the project a -// copy is routed to, so a namespace that gains them can publish what it holds. -func cellNamespace(_ context.Context, namespace *corev1.Namespace) []reconcile.Request { - return []reconcile.Request{{NamespacedName: client.ObjectKey{Name: namespace.Name}}} -} - // everyNamespace maps an object to one request that stands for every namespace. func everyNamespace[T client.Object](context.Context, T) []reconcile.Request { return []reconcile.Request{{}} @@ -541,8 +535,7 @@ func servesKind(mgr ctrl.Manager, obj client.Object) (bool, error) { } // reconcileNamespace republishes a namespace's slices when its reachability -// record or its namespace appears on the hub, or its project labels on the -// cell change. An empty name stands for every namespace. +// record on the hub changes. An empty name stands for every namespace. func (r *VPCEndpointSliceWriteBackReconciler) reconcileNamespace(ctx context.Context, req reconcile.Request) (ctrl.Result, error) { return ctrl.Result{}, r.republishNamespace(ctx, req.Name) } @@ -617,20 +610,6 @@ func (r *VPCEndpointSliceWriteBackReconciler) SetupWithManager(mgr mcmanager.Man return fmt.Errorf("unable to watch edge reachability on the hub: %w", err) } - if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). - WatchesRawSource(source.Kind( - r.HubCluster.GetCache(), - &corev1.Namespace{}, - handler.TypedEnqueueRequestsFromMapFunc(cellNamespace), - predicate.TypedFuncs[*corev1.Namespace]{ - CreateFunc: func(event.TypedCreateEvent[*corev1.Namespace]) bool { return true }, - }, - )). - Named("vpcendpointslice_writeback_hub_namespaces"). - Complete(reconcile.Func(r.reconcileNamespace)); err != nil { - return fmt.Errorf("unable to watch namespaces on the hub: %w", err) - } - servingLocations, err := servesKind(mgr.GetLocalManager(), &locationsv1alpha1.ServingLocation{}) if err != nil { return err @@ -648,18 +627,6 @@ func (r *VPCEndpointSliceWriteBackReconciler) SetupWithManager(mgr mcmanager.Man } } - if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). - WatchesRawSource(source.Kind( - mgr.GetLocalManager().GetCache(), - &corev1.Namespace{}, - handler.TypedEnqueueRequestsFromMapFunc(cellNamespace), - predicate.TypedLabelChangedPredicate[*corev1.Namespace]{}, - )). - Named("vpcendpointslice_writeback_namespaces"). - Complete(reconcile.Func(r.reconcileNamespace)); err != nil { - return fmt.Errorf("unable to watch the cell's namespaces: %w", err) - } - if err := ctrl.NewControllerManagedBy(mgr.GetLocalManager()). WatchesRawSource(source.Kind( r.HubCluster.GetCache(), diff --git a/internal/controller/vpcendpointslice_writeback_test.go b/internal/controller/vpcendpointslice_writeback_test.go index 700eb736..ba9c54d7 100644 --- a/internal/controller/vpcendpointslice_writeback_test.go +++ b/internal/controller/vpcendpointslice_writeback_test.go @@ -594,35 +594,3 @@ func TestACopyFromAnotherLocationIsLeftToItsOwnCell(t *testing.T) { require.Empty(t, r.writeBack.copySource(r.ctx, copied)) } - -// A slice can reach the cell before federation has made its hub namespace. -// The namespace appearing on the hub is what publishes it. -func TestASliceIsPublishedOnceItsHubNamespaceAppears(t *testing.T) { - r := newReachability(t) - - ns := &corev1.Namespace{} - ns.Name = r.namespace + "-late" - ns.Labels = map[string]string{ - downstreamclient.UpstreamOwnerNamespaceLabel: "proj-" + scenarioSuffix(t), - downstreamclient.UpstreamOwnerClusterNameLabel: "cluster-" + testProject, - } - require.NoError(t, r.cell.Create(r.ctx, ns)) - r.namespace = ns.Name - r.sliceOnCell() - - r.publish(liveSliceName) - _, found := r.hubCopy() - require.False(t, found) - - hubNamespace := &corev1.Namespace{} - hubNamespace.Name = ns.Name - require.NoError(t, r.hub.Create(r.ctx, hubNamespace)) - - requests := cellNamespace(r.ctx, hubNamespace) - require.Equal(t, []reconcile.Request{{NamespacedName: client.ObjectKey{Name: ns.Name}}}, requests) - _, err := r.writeBack.reconcileNamespace(r.ctx, requests[0]) - require.NoError(t, err) - - _, found = r.hubCopy() - require.True(t, found, "the hub namespace appearing publishes the slice") -} From 3b8b494fb5422fdbc2c4a0c59df2474c70f72500 Mon Sep 17 00:00:00 2001 From: Scot Schuchert-Wells Date: Tue, 10 Nov 2026 14:41:08 +0000 Subject: [PATCH 5/5] style: gofmt the edge reachability controller --- internal/controller/edgereachability_controller.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/controller/edgereachability_controller.go b/internal/controller/edgereachability_controller.go index 39b49a46..eb581344 100644 --- a/internal/controller/edgereachability_controller.go +++ b/internal/controller/edgereachability_controller.go @@ -16,8 +16,8 @@ import ( "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/event" + "sigs.k8s.io/controller-runtime/pkg/handler" "sigs.k8s.io/controller-runtime/pkg/log" "sigs.k8s.io/controller-runtime/pkg/predicate" mcbuilder "sigs.k8s.io/multicluster-runtime/pkg/builder"