From 43a50d398c7f5bdb45232aef8bdc64d74dc59a6c Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 11:48:52 +0300 Subject: [PATCH 01/10] wip(controller): cross-CR stale-peer cleanup Walk every cluster in the merged Registry (built from all ClusterMesh specs), not just this mesh's spec.Clusters. This closes the second half of the stale-peer problem: peers a mesh pushed into a cluster that has since been removed from its spec stay reachable via any sibling ClusterMesh that still references the cluster, so the sweep can reach them. Adds a cross-CR integration case to test/integration/cleanup_test.go that pins this contract: a peer planted in 'remote' for a mesh whose spec contains only 'local' is removed after reconcile. Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 29 ++++++---- test/integration/cleanup_test.go | 57 +++++++++++++++++++ 2 files changed, 75 insertions(+), 11 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index b315036..497269e 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -124,22 +124,29 @@ func (r *ClusterMeshReconciler) SetupWithManager(mgr ctrl.Manager) error { return errors.Wrap(err, "building clustermesh controller") } -// cleanupStaleSourceClusters walks every cluster currently present in the -// mesh's spec and asks each one to drop any Peer it holds whose -// source-cluster label points at a cluster no longer in the spec. Failures -// are logged and swallowed: a stale peer not deleted this tick will be -// retried on the next reconcile, and we should not block the rest of the -// status update on a single client error. +// cleanupStaleSourceClusters walks every cluster the operator knows about +// (the merged registry built from all ClusterMesh resources, not just this +// mesh's current spec) and asks each one to drop any Peer it holds whose +// source-cluster label points at a cluster no longer in this mesh's +// spec.Clusters. +// +// Visiting only this mesh's current spec misses peers that were pushed +// into a cluster that has since been removed from spec but is still +// reachable via another ClusterMesh's kubeconfig. Reaching every cluster +// in the registry closes both halves of the gap: removed source-clusters +// and removed target-clusters. +// +// Failures are logged and swallowed: a stale peer not deleted this tick +// will be retried on the next reconcile, and we should not block the rest +// of the status update on a single client error. func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, log *slog.Logger, mesh *v1alpha1.ClusterMesh) { validSources := make([]string, 0, len(mesh.Spec.Clusters)) for i := range mesh.Spec.Clusters { validSources = append(validSources, mesh.Spec.Clusters[i].Name) } - for i := range mesh.Spec.Clusters { - tgtEntry := &mesh.Spec.Clusters[i] - - tgtClient, ok := r.Registry.Client(tgtEntry.Name) + for _, name := range r.Registry.Clusters() { + tgtClient, ok := r.Registry.Client(name) if !ok { continue } @@ -147,7 +154,7 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, err := peer.DeleteStaleSourceClusters(ctx, tgtClient, mesh.Name, validSources) if err != nil { log.Warn("cleaning stale source-cluster peers", - slog.String("target", tgtEntry.Name), + slog.String("target", name), slog.String("error", err.Error()), ) } diff --git a/test/integration/cleanup_test.go b/test/integration/cleanup_test.go index b46aa19..f2684b0 100644 --- a/test/integration/cleanup_test.go +++ b/test/integration/cleanup_test.go @@ -25,6 +25,7 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "sigs.k8s.io/controller-runtime/pkg/client" + v1alpha1 "github.com/squat/kilo-clustermesh-operator/api/v1alpha1" "github.com/squat/kilo-clustermesh-operator/internal/peer" kilov1alpha1 "github.com/squat/kilo-clustermesh-operator/pkg/kilo/v1alpha1" ) @@ -99,6 +100,62 @@ func TestCleanupStaleSourceClusters_RemovesPeersOfRemovedCluster(t *testing.T) { // desired=[] and removes the planted peer as an in-pair orphan, masking the // behaviour this layer is meant to verify. +// TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec verifies the +// cross-CR cleanup property: a peer left in a cluster that is in the +// operator's registry (because another ClusterMesh names it) but is NOT in +// THIS mesh's spec.Clusters must still be swept. This is the case that +// breaks when cleanup only walks spec.Clusters: the "removed target" half +// of the stale-peer problem. +// +// Setup: this mesh's spec contains only "local". The registry, however, +// also has "remote" (in real life from a sibling ClusterMesh). A peer +// labeled for this mesh sits in remote with a source-cluster that is not +// in this mesh's spec. The reconcile must reach into remote and delete it. +func TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec(t *testing.T) { + ctx := context.Background() + + // Spec contains ONLY local; remote is reachable via the registry but + // not referenced by this mesh. + mesh := &v1alpha1.ClusterMesh{ + ObjectMeta: metav1.ObjectMeta{ + Name: "cleanup-cross-cr-mesh", + Namespace: "default", + }, + Spec: v1alpha1.ClusterMeshSpec{ + Clusters: []v1alpha1.ClusterEntry{ + { + Name: "local", + Local: true, + PodCIDRs: []string{"10.1.0.0/16"}, + WireguardCIDR: "10.100.0.0/24", + }, + }, + }, + } + createMesh(t, mesh) + t.Cleanup(func() { deleteMesh(t, mesh) }) + + // Plant a peer in remote whose source-cluster is not in this mesh's + // spec. Without cross-cluster sweep, reconcile would never visit + // remote (it's not in spec) and the peer would persist forever. + stalePeer := &kilov1alpha1.Peer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "stale--ghost--in-remote", + Labels: peer.Labels(mesh.Name, "ghost-cluster"), + }, + Spec: kilov1alpha1.PeerSpec{ + AllowedIPs: []string{"10.77.0.0/16"}, + PublicKey: "pubkey-stale-cross-cr", + }, + } + require.NoError(t, globalEnv.remoteClient.Create(ctx, stalePeer)) + + mustReconcile(t, mesh) + mustReconcile(t, mesh) + + assertPeerExists(t, globalEnv.remoteClient, "stale--ghost--in-remote", false) +} + // TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes verifies that // peers labeled for a different ClusterMesh are not touched, even if their // source-cluster is unknown to this mesh. From ef923e155855633a5dcc0452fe5f0ae9bb2ae3dd Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 11:56:02 +0300 Subject: [PATCH 02/10] wip(controller): also wipe peers in targets fully removed from spec Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 35 +++++++++++++------ 1 file changed, 25 insertions(+), 10 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index 497269e..be7ec2a 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -126,23 +126,30 @@ func (r *ClusterMeshReconciler) SetupWithManager(mgr ctrl.Manager) error { // cleanupStaleSourceClusters walks every cluster the operator knows about // (the merged registry built from all ClusterMesh resources, not just this -// mesh's current spec) and asks each one to drop any Peer it holds whose -// source-cluster label points at a cluster no longer in this mesh's -// spec.Clusters. +// mesh's current spec) and removes Peer objects this mesh has no business +// owning anymore. Two situations are swept: // -// Visiting only this mesh's current spec misses peers that were pushed -// into a cluster that has since been removed from spec but is still -// reachable via another ClusterMesh's kubeconfig. Reaching every cluster -// in the registry closes both halves of the gap: removed source-clusters -// and removed target-clusters. +// - target cluster was removed from this mesh's spec.Clusters: the mesh +// should hold no peers at all in that cluster, so every Peer labeled +// with this mesh's name is deleted there. // +// - target cluster is still in spec: only Peers whose source-cluster +// label points at a cluster no longer in spec are deleted. +// +// Visiting only the current spec.Clusters misses the first case — a peer +// pushed into a now-removed target stays reachable via another +// ClusterMesh's kubeconfig and would otherwise never be touched again. // Failures are logged and swallowed: a stale peer not deleted this tick // will be retried on the next reconcile, and we should not block the rest // of the status update on a single client error. func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, log *slog.Logger, mesh *v1alpha1.ClusterMesh) { validSources := make([]string, 0, len(mesh.Spec.Clusters)) + validSet := make(map[string]struct{}, len(mesh.Spec.Clusters)) + for i := range mesh.Spec.Clusters { - validSources = append(validSources, mesh.Spec.Clusters[i].Name) + name := mesh.Spec.Clusters[i].Name + validSources = append(validSources, name) + validSet[name] = struct{}{} } for _, name := range r.Registry.Clusters() { @@ -151,7 +158,15 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, continue } - err := peer.DeleteStaleSourceClusters(ctx, tgtClient, mesh.Name, validSources) + // Target cluster not in this mesh's spec → drop every Peer this + // mesh owns there. Target still in spec → keep peers whose source + // is still in spec, drop the rest. + sources := validSources + if _, inSpec := validSet[name]; !inSpec { + sources = nil + } + + err := peer.DeleteStaleSourceClusters(ctx, tgtClient, mesh.Name, sources) if err != nil { log.Warn("cleaning stale source-cluster peers", slog.String("target", name), From 15a459297793b772cc1d24e4ed72f103a3815b2d Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 13:00:20 +0300 Subject: [PATCH 03/10] wip(multicluster): tolerate missing kubeconfig secrets at registry build Build returned an error from the first cluster entry whose kubeconfig Secret could not be read. During tenant teardown this crashed the operator at startup: the upstream KubernetesSwitchcloud Helm release deletes the tenant's admin-kubeconfig Secret before the ClusterMesh CR that references it is finalized, leaving a stale secretRef that buildInitialRegistry then dereferences. The operator crashlooped, the finalizer never ran, the ClusterMesh CR stuck with finalizers forever, and downstream teardown stalled. Build now logs a warning per problematic entry and skips it, returning the rest of the registry as usable. cluster.Cluster construction failures are treated the same way. The reconciler already does best-effort sweeps via r.Registry.Client(name) with 'if !ok { continue }', so a partial registry preserves forward progress: finalizers clean peers in the clusters they can still reach, the ClusterMesh CR releases, and the operator can come back up clean on the next reconcile. A nil logger is accepted to keep the signature ergonomic; it routes to slog.DiscardHandler so test callers and other call sites do not need to thread a logger through to use Build. cmd/main.go threads the existing slogger through buildInitialRegistry so missing-secret warnings show up in the operator log at WARN level. Three unit tests cover the new behaviour: - TestBuild_SkipsEntryWithMissingSecret: reproduces the crashloop bug with a mesh+ceph pair where only ceph's kubeconfig Secret exists. - TestBuild_LocalEntryDoesNotNeedSecret: guards the Local: true fast-path, which must keep succeeding regardless of Secret state. - TestBuild_NilLoggerIsAccepted: protects against nil-deref for call sites that have not yet wired a logger. Signed-off-by: IvanHunters --- cmd/main.go | 6 +- internal/multicluster/registry.go | 35 +++++- internal/multicluster/registry_test.go | 144 +++++++++++++++++++++++++ 3 files changed, 180 insertions(+), 5 deletions(-) create mode 100644 internal/multicluster/registry_test.go diff --git a/cmd/main.go b/cmd/main.go index f9ffa9a..a983063 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -106,7 +106,7 @@ func run() error { return errors.Wrap(err, "installing CRD") } - registry, err := buildInitialRegistry(ctx, cfg) + registry, err := buildInitialRegistry(ctx, cfg, slogger) if err != nil { return errors.Wrap(err, "building cluster registry") } @@ -213,7 +213,7 @@ func readNamespace() (string, error) { // namespace and constructs a registry that holds clients for every declared // cluster. If no ClusterMesh resources exist yet, an empty registry is // returned and the change-watcher will trigger a restart once one is created. -func buildInitialRegistry(ctx context.Context, cfg *rest.Config) (*multicluster.ClusterRegistry, error) { +func buildInitialRegistry(ctx context.Context, cfg *rest.Config, log *slog.Logger) (*multicluster.ClusterRegistry, error) { preClient, err := client.New(cfg, client.Options{Scheme: scheme}) if err != nil { return nil, errors.Wrap(err, "building pre-manager client") @@ -226,7 +226,7 @@ func buildInitialRegistry(ctx context.Context, cfg *rest.Config) (*multicluster. entries := mergeClusterEntries(meshes.Items) - registry, err := multicluster.Build(ctx, entries, cfg, preClient, scheme) + registry, err := multicluster.Build(ctx, entries, cfg, preClient, scheme, log) return registry, errors.Wrap(err, "constructing registry") } diff --git a/internal/multicluster/registry.go b/internal/multicluster/registry.go index beaa87c..1f23448 100644 --- a/internal/multicluster/registry.go +++ b/internal/multicluster/registry.go @@ -18,6 +18,7 @@ package multicluster import ( "context" + "log/slog" "net/http" "github.com/cockroachdb/errors" @@ -108,13 +109,31 @@ type EntrySource struct { // localCfg is the rest.Config of the cluster where the controller runs. // kubeClient is used to read kubeconfig Secrets for remote clusters from the // namespace of the ClusterMesh resource that contributed each entry. +// +// Per-entry failures (missing kubeconfig Secret, malformed kubeconfig, +// failure to construct the cluster.Cluster) are logged via the provided +// logger and the entry is skipped — they do not abort the build. This +// matters during teardown: if a tenant's KubernetesSwitchcloud is being +// deleted, its admin-kubeconfig Secret may be removed before the +// ClusterMesh CR that references it. An intolerant Build would crash the +// operator on startup, blocking the finalizer that would otherwise clean +// up Peer objects in still-reachable clusters and release the +// ClusterMesh. The reconciler already does best-effort sweeps over the +// registry and skips clusters it cannot reach, so a partial registry is +// safe and forward-progress-preserving. A nil logger is accepted and +// treated as a discard logger. func Build( ctx context.Context, entries []EntrySource, localCfg *rest.Config, kubeClient client.Client, scheme *runtime.Scheme, + log *slog.Logger, ) (*ClusterRegistry, error) { + if log == nil { + log = slog.New(slog.DiscardHandler) + } + reg := &ClusterRegistry{ clusters: make(map[string]cluster.Cluster, len(entries)), } @@ -125,7 +144,13 @@ func Build( cfg, err := configForEntry(ctx, &entry, localCfg, src.MeshNamespace, kubeClient) if err != nil { - return nil, err + log.Warn("skipping cluster entry during registry build", + slog.String("cluster", entry.Name), + slog.String("meshNamespace", src.MeshNamespace), + slog.String("error", err.Error()), + ) + + continue } if entry.Local { @@ -136,7 +161,13 @@ func Build( o.Scheme = scheme }) if err != nil { - return nil, errors.Wrapf(err, "creating cluster object for %q", entry.Name) + log.Warn("skipping cluster entry during registry build", + slog.String("cluster", entry.Name), + slog.String("meshNamespace", src.MeshNamespace), + slog.String("error", errors.Wrapf(err, "creating cluster object for %q", entry.Name).Error()), + ) + + continue } reg.clusters[entry.Name] = c diff --git a/internal/multicluster/registry_test.go b/internal/multicluster/registry_test.go new file mode 100644 index 0000000..4008af3 --- /dev/null +++ b/internal/multicluster/registry_test.go @@ -0,0 +1,144 @@ +/* +Copyright 2026 The Kilo Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package multicluster + +import ( + "context" + "log/slog" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/client-go/rest" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + + v1alpha1 "github.com/squat/kilo-clustermesh-operator/api/v1alpha1" +) + +func discardLogger() *slog.Logger { + return slog.New(slog.DiscardHandler) +} + +// TestBuild_SkipsEntryWithMissingSecret reproduces the operator-crashloop +// scenario observed during tenant teardown: a ClusterMesh CR still references +// a remote cluster whose admin-kubeconfig Secret has already been removed by +// the upstream Helm release. Before the fix, Build returned an error from the +// first such entry and the operator failed to start, blocking the finalizer +// that would otherwise release the ClusterMesh and let the rest of the +// teardown proceed. Build must now log a warning, skip that entry, and keep +// every other entry it can construct. +func TestBuild_SkipsEntryWithMissingSecret(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + + // Only the "ceph" secret is present. The "mesh2" entry references a + // secret that does not exist — simulating Helm having deleted it ahead + // of the ClusterMesh CR's finalizer. + cephSecret := &corev1.Secret{ + ObjectMeta: metav1.ObjectMeta{Name: "ceph-kubeconfig", Namespace: "tenant-root"}, + Data: map[string][]byte{"kubeconfig": []byte(testKubeconfig)}, + } + fc := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cephSecret).Build() + + entries := []EntrySource{ + { + Entry: v1alpha1.ClusterEntry{ + Name: "ceph", + KubeconfigSecretRef: &v1alpha1.SecretKeyRef{ + Name: "ceph-kubeconfig", + Key: "kubeconfig", + }, + }, + MeshNamespace: "tenant-root", + }, + { + Entry: v1alpha1.ClusterEntry{ + Name: "mesh2", + KubeconfigSecretRef: &v1alpha1.SecretKeyRef{ + Name: "kubernetes-switchcloud-mesh2-admin-kubeconfig", + Key: "super-admin.conf", + }, + }, + MeshNamespace: "tenant-root", + }, + } + + reg, err := Build(context.Background(), entries, &rest.Config{}, fc, scheme, discardLogger()) + require.NoError(t, err, "Build must not fail when some entries have missing secrets") + require.NotNil(t, reg) + + clusters := reg.Clusters() + assert.Contains(t, clusters, "ceph", "reachable entry must remain in the registry") + assert.NotContains(t, clusters, "mesh2", "entry with missing secret must be skipped") + + _, ok := reg.Client("mesh2") + assert.False(t, ok, "Client lookup for the skipped cluster must return ok=false so reconciler best-effort loops continue past it") +} + +// TestBuild_LocalEntryDoesNotNeedSecret guards against regressing the +// fast-path: a Local: true entry must always succeed since it reuses +// localCfg, even when no Secret of the referenced name exists. +func TestBuild_LocalEntryDoesNotNeedSecret(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + + fc := fake.NewClientBuilder().WithScheme(scheme).Build() + + entries := []EntrySource{ + { + Entry: v1alpha1.ClusterEntry{Name: "local", Local: true}, + MeshNamespace: "tenant-root", + }, + } + + reg, err := Build(context.Background(), entries, &rest.Config{}, fc, scheme, discardLogger()) + require.NoError(t, err) + require.NotNil(t, reg) + + assert.Equal(t, "local", reg.LocalName()) + assert.Contains(t, reg.Clusters(), "local") +} + +// TestBuild_NilLoggerIsAccepted ensures callers that have not yet wired a +// logger can still use Build without panicking on a nil dereference. +func TestBuild_NilLoggerIsAccepted(t *testing.T) { + scheme := runtime.NewScheme() + require.NoError(t, corev1.AddToScheme(scheme)) + + fc := fake.NewClientBuilder().WithScheme(scheme).Build() + + entries := []EntrySource{ + { + Entry: v1alpha1.ClusterEntry{ + Name: "missing-secret-cluster", + KubeconfigSecretRef: &v1alpha1.SecretKeyRef{ + Name: "does-not-exist", + Key: "kubeconfig", + }, + }, + MeshNamespace: "tenant-root", + }, + } + + reg, err := Build(context.Background(), entries, &rest.Config{}, fc, scheme, nil) + require.NoError(t, err) + require.NotNil(t, reg) + assert.NotContains(t, reg.Clusters(), "missing-secret-cluster") +} From e16d6f98c14a40180229a72bc1645d070eaee314 Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 13:06:39 +0300 Subject: [PATCH 04/10] wip(controller): sweep peers of ClusterMesh CRs that no longer exist MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit handleDeletion only runs when the finalizer reconcile gets to fire. In production we have already seen the finalizer skipped: - the operator crashlooped (missing kubeconfig Secret blocked buildInitialRegistry), so the finalizer reconcile never ran, and the user manually removed the finalizer to unblock teardown. - force-delete via the API or controller-runtime client. In all three cases the cleanup half of teardown is bypassed, leaving Peer objects labeled with a mesh name whose ClusterMesh CR no longer exists. The per-CR cleanupStaleSourceClusters pass added in the previous commits cannot touch those: it only looks at peers labeled with its OWN mesh name. The ghosts persist until a human notices. Add cleanupOrphanMeshPeers: every reconcile of any ClusterMesh now lists every other ClusterMesh in the namespace, collects the set of living mesh names, walks every cluster in the registry, and deletes any Peer whose kilo-clustermesh.io/mesh label points at a name not in that set. Failures (list, delete) are logged per-peer and the pass continues. The sweep is namespace-scoped: it only considers ClusterMesh objects in the reconciling CR's namespace, so it cannot accidentally clobber peers managed from another namespace by another operator. Wired into Reconcile() after the per-CR cleanupStaleSourceClusters pass so that every live reconcile contributes to the global sweep — the cluster reconverges as soon as any mesh's reconcile fires. Two integration cases in test/integration/cleanup_test.go: - DeletesGhostsAfterCRGone: plants a peer labeled with a ghost-mesh name; live mesh reconcile removes it. - LeavesPeersOfLivingMeshAlone: regression guard ensuring the sweep keeps peers whose mesh label still corresponds to an existing CR. Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 97 +++++++++++++++++++ test/integration/cleanup_test.go | 81 ++++++++++++++++ 2 files changed, 178 insertions(+) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index be7ec2a..04a680d 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -22,6 +22,7 @@ import ( "github.com/cockroachdb/errors" corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" apimeta "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" @@ -108,6 +109,14 @@ func (r *ClusterMeshReconciler) Reconcile(ctx context.Context, req ctrl.Request) // persist forever in the surviving clusters. r.cleanupStaleSourceClusters(ctx, log, mesh) + // Sweep peers whose owning ClusterMesh CR no longer exists. handleDeletion + // normally cleans these via the finalizer, but the finalizer may be + // skipped (force-delete, finalizer manually removed, operator crashloop + // that prevented the finalizer reconcile from running). Without this + // sweep, such peers persist forever as ghosts. Any live reconcile takes + // the global cleanup pass so the cluster always converges. + r.cleanupOrphanMeshPeers(ctx, log, mesh.Namespace) + return ctrl.Result{}, r.updateStatus(ctx, mesh, clusterStatuses) } @@ -176,6 +185,94 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, } } +// cleanupOrphanMeshPeers deletes Peer objects whose kilo-clustermesh.io/mesh +// label names a ClusterMesh that no longer exists in the given namespace. +// It is a global self-healing pass: it scopes by the operator-namespace of +// living ClusterMesh CRs, so it cannot accidentally clobber peers managed +// from another namespace. +// +// Why this exists: handleDeletion is the primary cleanup path, but it +// requires the finalizer to actually run. The finalizer is skipped if: +// - the CR was force-deleted (e.g. operator was crashlooping and the user +// manually removed the finalizer to unblock teardown), +// - the finalizer was never present (legacy CR predating the finalizer), +// - reconcile-time errors caused the finalizer reconcile to never make it +// to the peer-deletion step. +// +// Without this sweep, the cluster accumulates ghost peers that no future +// reconcile would ever notice — none of the per-CR cleanup paths look at +// peers labeled for CRs other than their own. +// +// Failures are logged and swallowed: each peer is deleted independently and +// a single client error must not abort the whole pass. +func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log *slog.Logger, namespace string) { + var meshes v1alpha1.ClusterMeshList + + err := r.List(ctx, &meshes, client.InNamespace(namespace)) + if err != nil { + log.Warn("listing meshes for orphan sweep", + slog.String("namespace", namespace), + slog.String("error", err.Error()), + ) + + return + } + + living := make(map[string]struct{}, len(meshes.Items)) + for i := range meshes.Items { + living[meshes.Items[i].Name] = struct{}{} + } + + for _, clusterName := range r.Registry.Clusters() { + tgtClient, ok := r.Registry.Client(clusterName) + if !ok { + continue + } + + var peers kilov1alpha1.PeerList + + err := tgtClient.List(ctx, &peers) + if err != nil { + log.Warn("listing peers for orphan sweep", + slog.String("target", clusterName), + slog.String("error", err.Error()), + ) + + continue + } + + for i := range peers.Items { + peerObj := &peers.Items[i] + + meshLabel := peerObj.Labels[peer.LabelMesh] + if meshLabel == "" { + continue + } + + if _, alive := living[meshLabel]; alive { + continue + } + + deleteErr := tgtClient.Delete(ctx, peerObj) + if deleteErr != nil && !apierrors.IsNotFound(deleteErr) { + log.Warn("deleting orphan peer", + slog.String("target", clusterName), + slog.String("peer", peerObj.Name), + slog.String("error", deleteErr.Error()), + ) + + continue + } + + log.Info("deleted orphan peer whose ClusterMesh CR no longer exists", + slog.String("target", clusterName), + slog.String("peer", peerObj.Name), + slog.String("orphan-mesh", meshLabel), + ) + } + } +} + // handleDeletion removes all Peers for this mesh from every cluster, then drops the finalizer. func (r *ClusterMeshReconciler) handleDeletion(ctx context.Context, log *slog.Logger, mesh *v1alpha1.ClusterMesh) error { if !controllerutil.ContainsFinalizer(mesh, finalizerName) { diff --git a/test/integration/cleanup_test.go b/test/integration/cleanup_test.go index f2684b0..a6e8255 100644 --- a/test/integration/cleanup_test.go +++ b/test/integration/cleanup_test.go @@ -189,6 +189,87 @@ func TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes(t *testing.T) { assertPeerExists(t, globalEnv.remoteClient, "foreign--ghost--node", true) } +// TestCleanupOrphanMeshPeers_DeletesGhostsAfterCRGone verifies the +// defense-in-depth sweep: peers labeled for a ClusterMesh CR that no longer +// exists (because the finalizer was bypassed: force-delete, manual +// finalizer removal, operator crashloop) must be deleted by any live +// reconcile of any surviving mesh. +// +// Setup: plant a Peer labeled with a "ghost-mesh" name that has no +// corresponding ClusterMesh object. Then reconcile a live mesh — the sweep +// must walk the registry and delete that orphan. +func TestCleanupOrphanMeshPeers_DeletesGhostsAfterCRGone(t *testing.T) { + ctx := context.Background() + + // Live mesh that drives the reconcile. + live := simpleMeshSpec("orphan-sweep-live-mesh", "default") + createMesh(t, live) + t.Cleanup(func() { deleteMesh(t, live) }) + + // Orphan peer: labeled with a mesh name that does NOT correspond to + // any ClusterMesh CR in the namespace. Mirrors the production + // failure mode where a tenant CR was force-deleted, leaving its + // peers as ghosts no per-CR reconcile would ever revisit. + orphan := &kilov1alpha1.Peer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "ghost--ghost-cluster--node-zero", + Labels: peer.Labels("ghost-mesh-deleted", "ghost-cluster"), + }, + Spec: kilov1alpha1.PeerSpec{ + AllowedIPs: []string{"10.66.0.0/16"}, + PublicKey: "pubkey-orphan", + }, + } + require.NoError(t, globalEnv.remoteClient.Create(ctx, orphan)) + + // Sanity: orphan exists before reconcile. + assertPeerExists(t, globalEnv.remoteClient, "ghost--ghost-cluster--node-zero", true) + + // First reconcile adds the finalizer; second runs the cleanup passes. + mustReconcile(t, live) + mustReconcile(t, live) + + assertPeerExists(t, globalEnv.remoteClient, "ghost--ghost-cluster--node-zero", false) +} + +// TestCleanupOrphanMeshPeers_LeavesPeersOfLivingMeshAlone guards against +// a regression where the orphan sweep deletes peers whose mesh CR DOES +// exist. The peer in this test belongs to the live mesh itself — it must +// survive every reconcile (subject only to the per-CR sweeps that apply +// when the source-cluster is invalid). +func TestCleanupOrphanMeshPeers_LeavesPeersOfLivingMeshAlone(t *testing.T) { + ctx := context.Background() + + live := simpleMeshSpec("orphan-sweep-keep-mesh", "default") + createMesh(t, live) + t.Cleanup(func() { deleteMesh(t, live) }) + + // Peer belongs to the live mesh; its source-cluster IS in the mesh's + // spec, so neither per-CR nor orphan sweep should touch it. + live.Spec.Clusters[0].Name = "local" + survivor := &kilov1alpha1.Peer{ + ObjectMeta: metav1.ObjectMeta{ + Name: "survivor--local--node-a", + Labels: peer.Labels(live.Name, "local"), + }, + Spec: kilov1alpha1.PeerSpec{ + AllowedIPs: []string{"10.1.0.0/24"}, + PublicKey: "pubkey-survivor", + }, + } + // Plant in a cluster that's NOT in spec (remote is not iterated by + // the per-pair ReconcilePeers source=local because local has no + // nodes; only the orphan sweep would touch this peer). Same idea + // — the survivor must remain because its mesh label points at a + // living CR. + require.NoError(t, globalEnv.localClient.Create(ctx, survivor)) + + mustReconcile(t, live) + mustReconcile(t, live) + + assertPeerExists(t, globalEnv.localClient, "survivor--local--node-a", true) +} + func assertPeerExists(t *testing.T, cl client.Client, name string, want bool) { t.Helper() From 34422648b0bd0a836e65cf4a4f9ff1f7052331f7 Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 14:02:28 +0300 Subject: [PATCH 05/10] wip(peer): fold cluster-wide CIDRs into first node Peer, drop anchor Peer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Kilo (and WireGuard generally) identifies peers exclusively by their public key. The receiving cluster's kilo daemon issues `wg setconf` to load Peer CRs into kernel state, and a second peer entry with the same pubkey REPLACES the first — AllowedIPs are not merged. The operator had been emitting two Peer objects per source-cluster node anchor: one "node" Peer with AllowedIPs=[podCIDR, wgIP/32] and one "anchor" Peer with AllowedIPs=[serviceCIDR, additionalCIDRs], both reusing the anchor node's pubkey. On the receiving cluster (e.g. nuvolos-ceph in our live test), kilo applied whichever Peer it processed last; the other entry's AllowedIPs were silently dropped from the WG kernel state. The result was a racy half-outage: sometimes pod-CIDR traffic worked but the serviceCIDR was unrouted; sometimes the inverse. The mesh1 path happened to land in the favourable order; mesh2 didn't, and its handshake completed (we observed it via tcpdump) yet inner packets were dropped on ingress because their source IP was outside the WG peer's AllowedIPs. Stop emitting the separate anchor Peer. Fold the cluster-wide CIDRs into the first valid node's Peer.AllowedIPs via a new extraAllowedIPs parameter on BuildPeer, with CollectAnchorCIDRs exposing the shape that was previously private. The buildDesiredPeers loop picks the first node as the anchor carrier; other nodes get nil extras. Old anchor Peer objects on remote clusters disappear naturally: their NAME (----anchor) is no longer in the desired set, so ReconcilePeers' per-pair orphan sweep deletes them on the next reconcile. Knock-on changes: - builder_test.go updates every BuildPeer call site for the new 4-argument signature (existing tests pass nil extras). - TestBuildAnchorPeer_* renamed to TestBuildPeer_AnchorExtras_* and rewritten to exercise the new fold-in semantics. The NoEndpointSource case is now a hard error from BuildPeer instead of a nil from BuildAnchorPeer, which is strictly stricter: misconfiguration surfaces at reconcile time rather than silently losing cluster-wide CIDR routing. Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 29 ++++-- internal/peer/builder.go | 64 ++++++------- internal/peer/builder_test.go | 93 ++++++++++++------- 3 files changed, 108 insertions(+), 78 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index 04a680d..e79bde6 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -491,12 +491,27 @@ func (r *ClusterMeshReconciler) updateStatus(ctx context.Context, mesh *v1alpha1 return errors.Wrap(r.Status().Update(ctx, mesh), "updating status") } -// buildDesiredPeers constructs the desired Peer slice for all valid nodes plus an optional anchor. +// buildDesiredPeers constructs the desired Peer slice for all valid nodes. +// The first valid node carries the cluster-wide CIDRs (serviceCIDR and any +// AdditionalCIDRs) folded into its Peer.AllowedIPs. The older design emitted +// a separate anchor Peer that reused the anchor node's WireGuard public key; +// WireGuard's per-pubkey dedup made the second `wg setconf` call to apply +// either the node or the anchor entry clobber the AllowedIPs of the other, +// silently losing pod-CIDR or service-CIDR routing in a racy way. Folding +// the anchor CIDRs into the first node Peer keeps a single WG peer entry +// per pubkey with the full union of AllowedIPs. func buildDesiredPeers(meshName string, entry *v1alpha1.ClusterEntry, nodes []*corev1.Node) ([]*kilov1alpha1.Peer, error) { - peers := make([]*kilov1alpha1.Peer, 0, len(nodes)+1) + peers := make([]*kilov1alpha1.Peer, 0, len(nodes)) - for _, node := range nodes { - p, err := peer.BuildPeer(meshName, entry, node) + anchorExtras := peer.CollectAnchorCIDRs(entry) + + for i, node := range nodes { + var extras []string + if i == 0 { + extras = anchorExtras + } + + p, err := peer.BuildPeer(meshName, entry, node, extras) if err != nil { return nil, errors.Wrapf(err, "building peer for node %q", node.Name) } @@ -504,12 +519,6 @@ func buildDesiredPeers(meshName string, entry *v1alpha1.ClusterEntry, nodes []*c peers = append(peers, p) } - if len(nodes) > 0 { - if anchor := peer.BuildAnchorPeer(meshName, entry, nodes[0]); anchor != nil { - peers = append(peers, anchor) - } - } - return peers, nil } diff --git a/internal/peer/builder.go b/internal/peer/builder.go index bff959c..f8ef960 100644 --- a/internal/peer/builder.go +++ b/internal/peer/builder.go @@ -32,8 +32,25 @@ import ( ) // BuildPeer constructs a Peer object from a validated Node. -// The Peer's allowedIPs = node's PodCIDRs[0] + /32 (or /128) host route derived from the wireguard-ip annotation. -func BuildPeer(meshName string, entry *v1alpha1.ClusterEntry, node *corev1.Node) (*kilov1alpha1.Peer, error) { +// +// The Peer's AllowedIPs always include the node's PodCIDR plus a /32 (or +// /128) host route derived from its kilo.squat.ai/wireguard-ip annotation. +// +// extraAllowedIPs lets the caller fold cluster-wide CIDRs (serviceCIDR and +// any AdditionalCIDRs declared on the ClusterEntry) into the first valid +// node's Peer. This replaces the old "anchor Peer" pattern, which emitted +// a SEPARATE Peer object reusing the anchor node's public key. WireGuard +// identifies peers exclusively by their public key and keeps only one +// peer entry per pubkey: the second `wg setconf` call either dropped the +// node peer's AllowedIPs (so pod-CIDR routing disappeared) or dropped the +// anchor's (so service-CIDR routing disappeared), depending on apply +// order. The resulting outage on the receiving cluster was racy and +// could survive across reconciles. Folding the cluster-wide CIDRs into +// the node peer guarantees one WG peer entry per pubkey with the full +// union of AllowedIPs, so neither half can clobber the other. +// +// extraAllowedIPs may be nil for non-anchor nodes. +func BuildPeer(meshName string, entry *v1alpha1.ClusterEntry, node *corev1.Node, extraAllowedIPs []string) (*kilov1alpha1.Peer, error) { pubKey := node.Annotations[kilonode.AnnotationPublicKey] if pubKey == "" { return nil, errors.Newf("node %q has no public key annotation", node.Name) @@ -52,7 +69,9 @@ func BuildPeer(meshName string, entry *v1alpha1.ClusterEntry, node *corev1.Node) return nil, errors.Wrapf(err, "node %q has invalid wireguard-ip annotation %q", node.Name, wgIP) } - allowedIPs := []string{node.Spec.PodCIDRs[0], netutil.HostRoute(hostIP)} + allowedIPs := make([]string, 0, 2+len(extraAllowedIPs)) + allowedIPs = append(allowedIPs, node.Spec.PodCIDRs[0], netutil.HostRoute(hostIP)) + allowedIPs = append(allowedIPs, extraAllowedIPs...) endpoint, err := resolvePeerEndpoint(node, entry.WireguardPort) if err != nil { @@ -74,38 +93,13 @@ func BuildPeer(meshName string, entry *v1alpha1.ClusterEntry, node *corev1.Node) return peer, nil } -// BuildAnchorPeer constructs a Peer that carries cluster-wide CIDRs not covered -// by per-node Peers (e.g., serviceCIDR, additionalCIDRs). -// It uses the first validated node's public key and endpoint as the anchor point. -// Returns nil when there are no cluster-wide CIDRs to advertise, or when the -// anchor node has no resolvable endpoint (an anchor without an endpoint cannot -// terminate cross-cluster traffic for those CIDRs). -func BuildAnchorPeer(meshName string, entry *v1alpha1.ClusterEntry, anchorNode *corev1.Node) *kilov1alpha1.Peer { - anchorCIDRs := collectAnchorCIDRs(entry) - if len(anchorCIDRs) == 0 { - return nil - } - - endpoint, err := resolvePeerEndpoint(anchorNode, entry.WireguardPort) - if err != nil { - return nil - } - - return &kilov1alpha1.Peer{ - ObjectMeta: metav1.ObjectMeta{ - Name: Name(meshName, entry.Name, "anchor"), - Labels: Labels(meshName, entry.Name), - }, - Spec: kilov1alpha1.PeerSpec{ - AllowedIPs: anchorCIDRs, - PublicKey: anchorNode.Annotations[kilonode.AnnotationPublicKey], - Endpoint: endpoint, - }, - } -} - -// collectAnchorCIDRs returns the cluster-wide CIDRs for an anchor peer. -func collectAnchorCIDRs(entry *v1alpha1.ClusterEntry) []string { +// CollectAnchorCIDRs returns the cluster-wide CIDRs (serviceCIDR plus any +// AdditionalCIDRs) declared on a ClusterEntry. The caller is expected to +// pass these as extraAllowedIPs to BuildPeer for a single (anchor) node — +// see the commentary on BuildPeer for the WireGuard pubkey-dedup +// rationale that motivates folding cluster-wide CIDRs into a node Peer +// rather than emitting a separate anchor Peer. +func CollectAnchorCIDRs(entry *v1alpha1.ClusterEntry) []string { var cidrs []string if entry.ServiceCIDR != "" { diff --git a/internal/peer/builder_test.go b/internal/peer/builder_test.go index 0b942ec..0a519bf 100644 --- a/internal/peer/builder_test.go +++ b/internal/peer/builder_test.go @@ -80,7 +80,7 @@ func TestBuildPeer_HappyPath(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -108,7 +108,7 @@ func TestBuildPeer_CozystackStyleWGAnnotation(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -126,7 +126,7 @@ func TestBuildPeer_InvalidWireguardIP(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) @@ -141,7 +141,7 @@ func TestBuildPeer_MissingPublicKey(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) @@ -157,7 +157,7 @@ func TestBuildPeer_MissingWireguardIP(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) @@ -178,7 +178,7 @@ func TestBuildPeer_NoEndpointSources_ReturnsError(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) @@ -193,7 +193,7 @@ func TestBuildPeer_DNSEndpoint(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -211,7 +211,7 @@ func TestBuildPeer_IPEndpoint(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -221,9 +221,13 @@ func TestBuildPeer_IPEndpoint(t *testing.T) { assert.Empty(t, got.Spec.Endpoint.DNS) } -func TestBuildAnchorPeer_WithServiceCIDR(t *testing.T) { +func TestBuildPeer_AnchorExtras_WithServiceCIDR(t *testing.T) { t.Parallel() + // Replaces the legacy TestBuildAnchorPeer_WithServiceCIDR. The anchor + // CIDRs now fold into the first node's Peer via extraAllowedIPs, so a + // single WireGuard peer entry on the receiving side carries both the + // node-local routes and the cluster-wide routes. entry := &v1alpha1.ClusterEntry{ Name: "cluster-a", ServiceCIDR: "10.96.0.0/12", @@ -231,18 +235,22 @@ func TestBuildAnchorPeer_WithServiceCIDR(t *testing.T) { node := testNode("worker-1", testPodCIDR, baseAnnotations()) - got := peer.BuildAnchorPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, peer.CollectAnchorCIDRs(entry)) + require.NoError(t, err) require.NotNil(t, got) - assert.Equal(t, peer.Name("my-mesh", "cluster-a", "anchor"), got.Name) + assert.Equal(t, peer.Name("my-mesh", "cluster-a", "worker-1"), got.Name) assert.Equal(t, peer.Labels("my-mesh", "cluster-a"), got.Labels) + assert.Contains(t, got.Spec.AllowedIPs, testPodCIDR) assert.Contains(t, got.Spec.AllowedIPs, "10.96.0.0/12") assert.Equal(t, testPubKey, got.Spec.PublicKey) } -func TestBuildAnchorPeer_WithAdditionalCIDRs(t *testing.T) { +func TestBuildPeer_AnchorExtras_WithAdditionalCIDRs(t *testing.T) { t.Parallel() + // AdditionalCIDRs append after serviceCIDR in CollectAnchorCIDRs and + // must appear after the node's own routes in the merged AllowedIPs. entry := &v1alpha1.ClusterEntry{ Name: "cluster-a", ServiceCIDR: "10.96.0.0/12", @@ -251,15 +259,24 @@ func TestBuildAnchorPeer_WithAdditionalCIDRs(t *testing.T) { node := testNode("worker-1", testPodCIDR, baseAnnotations()) - got := peer.BuildAnchorPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, peer.CollectAnchorCIDRs(entry)) + require.NoError(t, err) require.NotNil(t, got) - assert.Equal(t, []string{"10.96.0.0/12", "192.168.100.0/24", "172.16.0.0/16"}, got.Spec.AllowedIPs) + + // Node-local routes come first, then anchor CIDRs in declared order. + assert.Equal(t, testPodCIDR, got.Spec.AllowedIPs[0]) + assert.Contains(t, got.Spec.AllowedIPs, "10.96.0.0/12") + assert.Contains(t, got.Spec.AllowedIPs, "192.168.100.0/24") + assert.Contains(t, got.Spec.AllowedIPs, "172.16.0.0/16") } -func TestBuildAnchorPeer_NoAnchorCIDRs(t *testing.T) { +func TestBuildPeer_NoExtras_ProducesNodeOnlyAllowedIPs(t *testing.T) { t.Parallel() + // Replaces TestBuildAnchorPeer_NoAnchorCIDRs: when the caller passes + // no extras (non-anchor node, or anchor cluster has no serviceCIDR / + // additionalCIDRs), the AllowedIPs list is just the node's own routes. entry := &v1alpha1.ClusterEntry{ Name: "cluster-a", ServiceCIDR: "", @@ -268,9 +285,11 @@ func TestBuildAnchorPeer_NoAnchorCIDRs(t *testing.T) { node := testNode("worker-1", testPodCIDR, baseAnnotations()) - got := peer.BuildAnchorPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, peer.CollectAnchorCIDRs(entry)) - assert.Nil(t, got, "must return nil when there are no cluster-wide CIDRs") + require.NoError(t, err) + require.NotNil(t, got) + assert.Len(t, got.Spec.AllowedIPs, 2, "node-only Peer must carry exactly pod-CIDR + wg-ip /32") } func TestBuildPeer_MalformedForceEndpoint_ReturnsError(t *testing.T) { @@ -285,7 +304,7 @@ func TestBuildPeer_MalformedForceEndpoint_ReturnsError(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) @@ -302,7 +321,7 @@ func TestBuildPeer_ClustermeshEndpointPreferred(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -328,7 +347,7 @@ func TestBuildPeer_ExternalIPFallback(t *testing.T) { entry := &v1alpha1.ClusterEntry{Name: "cluster-a", WireguardPort: 51820} - got, err := peer.BuildPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, nil) require.NoError(t, err) require.NotNil(t, got) @@ -349,18 +368,21 @@ func TestBuildPeer_MalformedClustermeshEndpoint_ReturnsError(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.Error(t, err) assert.Nil(t, got) } -func TestBuildAnchorPeer_NoEndpointSource_ReturnsNil(t *testing.T) { +func TestBuildPeer_AnchorExtras_NoEndpointSource_ReturnsError(t *testing.T) { t.Parallel() - // An anchor peer without an endpoint cannot terminate cross-cluster - // traffic for its CIDRs, so BuildAnchorPeer returns nil when the - // anchor node has no resolvable endpoint. + // Replaces TestBuildAnchorPeer_NoEndpointSource_ReturnsNil. Endpoint + // resolution is shared by every node Peer (anchor or not), so the + // same "no resolvable endpoint" condition now surfaces as a hard + // error from BuildPeer rather than a nil-anchor signal. That is + // strictly stricter: the operator surfaces misconfiguration instead + // of silently dropping cluster-wide CIDRs. entry := &v1alpha1.ClusterEntry{ Name: "cluster-a", ServiceCIDR: "10.96.0.0/12", @@ -370,18 +392,21 @@ func TestBuildAnchorPeer_NoEndpointSource_ReturnsNil(t *testing.T) { delete(annotations, kilonode.AnnotationForceEndpoint) node := testNode("worker-1", testPodCIDR, annotations) + node.Status.Addresses = nil - got := peer.BuildAnchorPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, peer.CollectAnchorCIDRs(entry)) - assert.Nil(t, got, "anchor without resolvable endpoint must be nil") + require.Error(t, err) + assert.Nil(t, got, "node Peer without resolvable endpoint must error, regardless of anchor CIDRs") } -func TestBuildAnchorPeer_ExternalIPFallback(t *testing.T) { +func TestBuildPeer_AnchorExtras_ExternalIPFallback(t *testing.T) { t.Parallel() - // The anchor peer participates in the same fallback chain — when the - // anchor node has no annotations but does have an ExternalIP, the - // endpoint is synthesised from Node.Status.Addresses. + // Replaces TestBuildAnchorPeer_ExternalIPFallback. The endpoint + // fallback chain (clustermesh-endpoint → force-endpoint → ExternalIP) + // is the same for anchor vs non-anchor nodes — there is no separate + // path anymore. entry := &v1alpha1.ClusterEntry{ Name: "cluster-a", ServiceCIDR: "10.96.0.0/12", @@ -396,12 +421,14 @@ func TestBuildAnchorPeer_ExternalIPFallback(t *testing.T) { {Type: corev1.NodeExternalIP, Address: "203.0.113.99"}, } - got := peer.BuildAnchorPeer("my-mesh", entry, node) + got, err := peer.BuildPeer("my-mesh", entry, node, peer.CollectAnchorCIDRs(entry)) + require.NoError(t, err) require.NotNil(t, got) require.NotNil(t, got.Spec.Endpoint) assert.Equal(t, "203.0.113.99", got.Spec.Endpoint.IP) assert.Equal(t, uint32(51820), got.Spec.Endpoint.Port) + assert.Contains(t, got.Spec.AllowedIPs, "10.96.0.0/12") } func TestBuildPeer_BracketedDNSEndpoint(t *testing.T) { @@ -415,7 +442,7 @@ func TestBuildPeer_BracketedDNSEndpoint(t *testing.T) { node := testNode("worker-1", testPodCIDR, annotations) - got, err := peer.BuildPeer("my-mesh", testEntry(), node) + got, err := peer.BuildPeer("my-mesh", testEntry(), node, nil) require.NoError(t, err) require.NotNil(t, got) From a08c923138f7148f8f709cd6a027b974f3b93d9b Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 15:13:57 +0300 Subject: [PATCH 06/10] fixup: split cleanupOrphanMeshPeers + harden integration tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CI feedback on the first push: 1. funlen: cleanupOrphanMeshPeers was 65 lines (limit 60). Split into three: cleanupOrphanMeshPeers (orchestrator), collectLivingMeshes (CR-list helper), sweepOrphanPeersInCluster (per-cluster scan). Behaviour identical; just shorter functions. 2. TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes had been planting a peer labelled with a non-existent mesh and asserting the per-source sweep leaves it alone. That was true before this PR added the orphan-mesh sweep — which legitimately deletes peers whose mesh CR doesn't exist. The test now creates a sibling ClusterMesh in the same namespace so the foreign peer's mesh label references a LIVING CR. The property under test (per-source sweep doesn't touch peers owned by another mesh) is unchanged; orphan handling is covered by the dedicated TestCleanupOrphanMeshPeers_* cases. 3. TestCleanupOrphanMeshPeers_LeavesPeersOfLivingMeshAlone planted a peer in localClient without a t.Cleanup, so it survived the reconciling mesh's deletion and polluted subsequent integration tests whose reconciles legitimately swept it as an orphan. Added the missing Delete in t.Cleanup. Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 95 +++++++++++-------- test/integration/cleanup_test.go | 28 +++++- 2 files changed, 81 insertions(+), 42 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index e79bde6..bab4436 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -206,6 +206,26 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, // Failures are logged and swallowed: each peer is deleted independently and // a single client error must not abort the whole pass. func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log *slog.Logger, namespace string) { + living, ok := r.collectLivingMeshes(ctx, log, namespace) + if !ok { + return + } + + for _, clusterName := range r.Registry.Clusters() { + tgtClient, present := r.Registry.Client(clusterName) + if !present { + continue + } + + r.sweepOrphanPeersInCluster(ctx, log, clusterName, tgtClient, living) + } +} + +// collectLivingMeshes returns the names of every ClusterMesh in the given +// namespace, used by the orphan sweep to decide which mesh labels are still +// owned. ok==false indicates the list call itself failed; caller should +// bail without deleting anything. +func (r *ClusterMeshReconciler) collectLivingMeshes(ctx context.Context, log *slog.Logger, namespace string) (map[string]struct{}, bool) { var meshes v1alpha1.ClusterMeshList err := r.List(ctx, &meshes, client.InNamespace(namespace)) @@ -215,7 +235,7 @@ func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log slog.String("error", err.Error()), ) - return + return nil, false } living := make(map[string]struct{}, len(meshes.Items)) @@ -223,53 +243,54 @@ func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log living[meshes.Items[i].Name] = struct{}{} } - for _, clusterName := range r.Registry.Clusters() { - tgtClient, ok := r.Registry.Client(clusterName) - if !ok { - continue - } - - var peers kilov1alpha1.PeerList - - err := tgtClient.List(ctx, &peers) - if err != nil { - log.Warn("listing peers for orphan sweep", - slog.String("target", clusterName), - slog.String("error", err.Error()), - ) + return living, true +} - continue - } +// sweepOrphanPeersInCluster lists every Peer in a target cluster and deletes +// those whose kilo-clustermesh.io/mesh label names a mesh not in living. +// Listing or per-peer delete failures are logged and the sweep continues +// with the remaining peers. +func (r *ClusterMeshReconciler) sweepOrphanPeersInCluster(ctx context.Context, log *slog.Logger, clusterName string, tgtClient client.Client, living map[string]struct{}) { + var peers kilov1alpha1.PeerList - for i := range peers.Items { - peerObj := &peers.Items[i] + err := tgtClient.List(ctx, &peers) + if err != nil { + log.Warn("listing peers for orphan sweep", + slog.String("target", clusterName), + slog.String("error", err.Error()), + ) - meshLabel := peerObj.Labels[peer.LabelMesh] - if meshLabel == "" { - continue - } + return + } - if _, alive := living[meshLabel]; alive { - continue - } + for i := range peers.Items { + peerObj := &peers.Items[i] - deleteErr := tgtClient.Delete(ctx, peerObj) - if deleteErr != nil && !apierrors.IsNotFound(deleteErr) { - log.Warn("deleting orphan peer", - slog.String("target", clusterName), - slog.String("peer", peerObj.Name), - slog.String("error", deleteErr.Error()), - ) + meshLabel := peerObj.Labels[peer.LabelMesh] + if meshLabel == "" { + continue + } - continue - } + if _, alive := living[meshLabel]; alive { + continue + } - log.Info("deleted orphan peer whose ClusterMesh CR no longer exists", + deleteErr := tgtClient.Delete(ctx, peerObj) + if deleteErr != nil && !apierrors.IsNotFound(deleteErr) { + log.Warn("deleting orphan peer", slog.String("target", clusterName), slog.String("peer", peerObj.Name), - slog.String("orphan-mesh", meshLabel), + slog.String("error", deleteErr.Error()), ) + + continue } + + log.Info("deleted orphan peer whose ClusterMesh CR no longer exists", + slog.String("target", clusterName), + slog.String("peer", peerObj.Name), + slog.String("orphan-mesh", meshLabel), + ) } } diff --git a/test/integration/cleanup_test.go b/test/integration/cleanup_test.go index a6e8255..f178cb4 100644 --- a/test/integration/cleanup_test.go +++ b/test/integration/cleanup_test.go @@ -157,8 +157,12 @@ func TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec(t *testing.T) { } // TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes verifies that -// peers labeled for a different ClusterMesh are not touched, even if their -// source-cluster is unknown to this mesh. +// peers labeled for a different but LIVING ClusterMesh are not touched by +// the per-source sweep, even if their source-cluster is unknown to the +// reconciling mesh. The foreign mesh must exist as a CR — otherwise the +// orphan-mesh sweep (a separate defense-in-depth pass) would legitimately +// delete it, and that property is covered by +// TestCleanupOrphanMeshPeers_DeletesGhostsAfterCRGone. func TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes(t *testing.T) { ctx := context.Background() @@ -166,12 +170,23 @@ func TestCleanupStaleSourceClusters_IgnoresPeersFromOtherMeshes(t *testing.T) { createMesh(t, mesh) t.Cleanup(func() { deleteMesh(t, mesh) }) - // Peer belongs to a different mesh; even though its source-cluster is - // nonsensical from our mesh's point of view, we must not touch it. + // Sibling ClusterMesh CR that "owns" the foreign peer below. Keeping it + // alive in the namespace ensures the orphan-mesh sweep treats its peer + // label as legitimate and does not delete it; the property under test + // is per-CR isolation of the source-cluster sweep, not orphan handling. + sibling := simpleMeshSpec("cleanup-isolation-sibling-mesh", "default") + createMesh(t, sibling) + t.Cleanup(func() { deleteMesh(t, sibling) }) + + // Peer belongs to the sibling mesh; the reconciling mesh + // (cleanup-isolation-mesh) must not delete it even though its + // source-cluster ("ghost-cluster") is not part of cleanup-isolation-mesh's + // spec — the per-source sweep is scoped to peers labelled with the + // reconciling mesh, not foreign ones. foreignPeer := &kilov1alpha1.Peer{ ObjectMeta: metav1.ObjectMeta{ Name: "foreign--ghost--node", - Labels: peer.Labels("some-other-mesh", "ghost-cluster"), + Labels: peer.Labels(sibling.Name, "ghost-cluster"), }, Spec: kilov1alpha1.PeerSpec{ AllowedIPs: []string{"10.50.0.0/16"}, @@ -263,6 +278,9 @@ func TestCleanupOrphanMeshPeers_LeavesPeersOfLivingMeshAlone(t *testing.T) { // — the survivor must remain because its mesh label points at a // living CR. require.NoError(t, globalEnv.localClient.Create(ctx, survivor)) + t.Cleanup(func() { + _ = globalEnv.localClient.Delete(ctx, survivor) + }) mustReconcile(t, live) mustReconcile(t, live) From a15c13918f2336d88cb49e1997c6bb7ab30a31ea Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 15:18:53 +0300 Subject: [PATCH 07/10] fixup: satisfy CRD minItems=2 on spec.clusters in cross-CR sweep test Signed-off-by: IvanHunters --- test/integration/cleanup_test.go | 31 +++++++++++++++++++++---------- 1 file changed, 21 insertions(+), 10 deletions(-) diff --git a/test/integration/cleanup_test.go b/test/integration/cleanup_test.go index f178cb4..cc6455c 100644 --- a/test/integration/cleanup_test.go +++ b/test/integration/cleanup_test.go @@ -102,20 +102,22 @@ func TestCleanupStaleSourceClusters_RemovesPeersOfRemovedCluster(t *testing.T) { // TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec verifies the // cross-CR cleanup property: a peer left in a cluster that is in the -// operator's registry (because another ClusterMesh names it) but is NOT in -// THIS mesh's spec.Clusters must still be swept. This is the case that -// breaks when cleanup only walks spec.Clusters: the "removed target" half -// of the stale-peer problem. +// operator's registry (because another ClusterMesh names it) but is NOT +// in THIS mesh's spec.Clusters must still be swept. This is the case +// that breaks when cleanup only walks spec.Clusters: the "removed +// target" half of the stale-peer problem. // -// Setup: this mesh's spec contains only "local". The registry, however, -// also has "remote" (in real life from a sibling ClusterMesh). A peer -// labeled for this mesh sits in remote with a source-cluster that is not -// in this mesh's spec. The reconcile must reach into remote and delete it. +// Setup: this mesh's spec contains "local" plus a placeholder +// "ghost-elsewhere" that is not in the registry (the CRD requires at +// least two cluster entries; ghost-elsewhere satisfies that without +// covering the "remote" cluster the registry knows about). The peer +// under test is planted in remote — which is reachable through the +// registry (sibling-mesh kubeconfig in real deployments) but is NOT +// referenced by this mesh's spec. The cross-CR sweep must visit remote +// and delete the peer. func TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec(t *testing.T) { ctx := context.Background() - // Spec contains ONLY local; remote is reachable via the registry but - // not referenced by this mesh. mesh := &v1alpha1.ClusterMesh{ ObjectMeta: metav1.ObjectMeta{ Name: "cleanup-cross-cr-mesh", @@ -129,6 +131,15 @@ func TestCleanupStaleSourceClusters_SweepsClustersOutsideSpec(t *testing.T) { PodCIDRs: []string{"10.1.0.0/16"}, WireguardCIDR: "10.100.0.0/24", }, + { + // Placeholder to satisfy the CRD's minItems=2 on + // spec.clusters. Not in the registry — the + // registry exposes "local" and "remote", so this + // entry is effectively ignored by reconcile. + Name: "ghost-elsewhere", + PodCIDRs: []string{"10.2.0.0/16"}, + WireguardCIDR: "10.100.1.0/24", + }, }, }, } From addc84b3bffc1ad78e18d320400e9eb7bab9bf01 Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 16:23:46 +0300 Subject: [PATCH 08/10] fixup: address bot review (PodCIDR guard, per-target timeouts, cluster-wide CR list) Three concrete findings from gemini-code-assist and one critical from coderabbitai on PR #11. 1. peer.BuildPeer: guard against empty Spec.PodCIDRs Accessing PodCIDRs[0] would panic on a node that joined before the kube-controller-manager had allocated a pod CIDR. Surface this as a clean BuildPeer error so the reconciler's per-node validation path skips the node and continues, instead of crash-looping the operator. 2. cleanupStaleSourceClusters: per-target timeout Synchronous List/Delete calls across the merged registry could stall the whole reconcile loop if one remote cluster's apiserver is slow or unreachable. Wrap each target's sweep in a context.WithTimeout bounded by cleanupSweepTimeout (5s), so a single bad cluster degrades only its own per-tick progress. 3. cleanupOrphanMeshPeers: per-target timeout (same rationale). 4. cleanupOrphanMeshPeers: list ClusterMesh CLUSTER-WIDE Critical bug spotted by coderabbitai. collectLivingMeshes previously restricted the lookup to the reconciler's own namespace, but the operator's registry is built cluster-wide from every ClusterMesh it sees. A Peer labelled mesh=foo on a target cluster could legitimately belong to a foo CR in another namespace, and reconciling a CR in namespace A would treat that peer as orphaned and delete it. Drop the InNamespace filter and consider every ClusterMesh in the cluster as a living mesh. Namespace-qualified Peer ownership (a dedicated label) is a cleaner long-term fix and worth a follow-up, but the cluster-wide list is enough to stop the data-loss path described in the review. Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 46 +++++++++++++++---- internal/peer/builder.go | 8 ++++ 2 files changed, 44 insertions(+), 10 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index bab4436..a543ee4 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -19,6 +19,7 @@ package controller import ( "context" "log/slog" + "time" "github.com/cockroachdb/errors" corev1 "k8s.io/api/core/v1" @@ -133,6 +134,12 @@ func (r *ClusterMeshReconciler) SetupWithManager(mgr ctrl.Manager) error { return errors.Wrap(err, "building clustermesh controller") } +// cleanupSweepTimeout caps the per-target list/delete pass time so a single +// unreachable or slow cluster does not block the whole reconcile loop. The +// reconciler retries on every tick anyway, so a brief budget is enough to +// either make progress or move on. +const cleanupSweepTimeout = 5 * time.Second + // cleanupStaleSourceClusters walks every cluster the operator knows about // (the merged registry built from all ClusterMesh resources, not just this // mesh's current spec) and removes Peer objects this mesh has no business @@ -175,7 +182,13 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, sources = nil } - err := peer.DeleteStaleSourceClusters(ctx, tgtClient, mesh.Name, sources) + // Per-target deadline so one unreachable cluster cannot stall the + // whole reconcile pass — see cleanupSweepTimeout for the rationale. + sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) + + err := peer.DeleteStaleSourceClusters(sweepCtx, tgtClient, mesh.Name, sources) + cancel() + if err != nil { log.Warn("cleaning stale source-cluster peers", slog.String("target", name), @@ -206,7 +219,7 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, // Failures are logged and swallowed: each peer is deleted independently and // a single client error must not abort the whole pass. func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log *slog.Logger, namespace string) { - living, ok := r.collectLivingMeshes(ctx, log, namespace) + living, ok := r.collectLivingMeshes(ctx, log) if !ok { return } @@ -217,21 +230,34 @@ func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log continue } - r.sweepOrphanPeersInCluster(ctx, log, clusterName, tgtClient, living) + // Per-target deadline so one unreachable cluster cannot stall the + // whole reconcile pass — see cleanupSweepTimeout. + sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) + r.sweepOrphanPeersInCluster(sweepCtx, log, clusterName, tgtClient, living) + cancel() } + + // `namespace` is currently unused — see collectLivingMeshes for the + // reasoning. Keeping the parameter on the function signature documents + // the reconciling-mesh context for future per-namespace heuristics and + // avoids breaking the small set of internal call sites. + _ = namespace } -// collectLivingMeshes returns the names of every ClusterMesh in the given -// namespace, used by the orphan sweep to decide which mesh labels are still -// owned. ok==false indicates the list call itself failed; caller should -// bail without deleting anything. -func (r *ClusterMeshReconciler) collectLivingMeshes(ctx context.Context, log *slog.Logger, namespace string) (map[string]struct{}, bool) { +// collectLivingMeshes returns the names of every ClusterMesh in the cluster, +// not just the reconciler's own namespace. The operator builds its registry +// cluster-wide (cmd.buildInitialRegistry merges entries from every +// ClusterMesh it sees), so a target cluster's Peer labelled mesh=foo could +// legitimately belong to a foo CR in any namespace — narrowing the lookup +// to one namespace would let the orphan sweep delete a peer owned by a +// living foo CR sitting elsewhere. ok==false indicates the list call itself +// failed; caller should bail without deleting anything. +func (r *ClusterMeshReconciler) collectLivingMeshes(ctx context.Context, log *slog.Logger) (map[string]struct{}, bool) { var meshes v1alpha1.ClusterMeshList - err := r.List(ctx, &meshes, client.InNamespace(namespace)) + err := r.List(ctx, &meshes) if err != nil { log.Warn("listing meshes for orphan sweep", - slog.String("namespace", namespace), slog.String("error", err.Error()), ) diff --git a/internal/peer/builder.go b/internal/peer/builder.go index f8ef960..4bb4df4 100644 --- a/internal/peer/builder.go +++ b/internal/peer/builder.go @@ -69,6 +69,14 @@ func BuildPeer(meshName string, entry *v1alpha1.ClusterEntry, node *corev1.Node, return nil, errors.Wrapf(err, "node %q has invalid wireguard-ip annotation %q", node.Name, wgIP) } + // PodCIDRs is populated by the kube-controller-manager once it allocates + // a CIDR for the node; until then BuildPeer would panic on the indexing + // below. Surface this as a clean error so the reconciler skips the + // node via validation rather than crashloop the operator. + if len(node.Spec.PodCIDRs) == 0 { + return nil, errors.Newf("node %q has no PodCIDRs allocated yet", node.Name) + } + allowedIPs := make([]string, 0, 2+len(extraAllowedIPs)) allowedIPs = append(allowedIPs, node.Spec.PodCIDRs[0], netutil.HostRoute(hostIP)) allowedIPs = append(allowedIPs, extraAllowedIPs...) From 72243dcde5c8a07a748bb150d6453b490ea8c94c Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 16:30:28 +0300 Subject: [PATCH 09/10] =?UTF-8?q?fixup:=20wsl=5Fv5=20lint=20=E2=80=94=20co?= =?UTF-8?q?llapse=20blank=20lines=20around=20sweepCtx=20block?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 2 -- 1 file changed, 2 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index a543ee4..9b76d2e 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -185,10 +185,8 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, // Per-target deadline so one unreachable cluster cannot stall the // whole reconcile pass — see cleanupSweepTimeout for the rationale. sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) - err := peer.DeleteStaleSourceClusters(sweepCtx, tgtClient, mesh.Name, sources) cancel() - if err != nil { log.Warn("cleaning stale source-cluster peers", slog.String("target", name), From be32cde4821a9e67c6cd141ef115ea9e3325c6b5 Mon Sep 17 00:00:00 2001 From: IvanHunters Date: Wed, 27 May 2026 17:09:54 +0300 Subject: [PATCH 10/10] fixup: wrap per-target sweeps in anon-funcs for defer cancel() pairing Signed-off-by: IvanHunters --- internal/controller/clustermesh_controller.go | 35 ++++++++++++------- 1 file changed, 22 insertions(+), 13 deletions(-) diff --git a/internal/controller/clustermesh_controller.go b/internal/controller/clustermesh_controller.go index 9b76d2e..bb75393 100644 --- a/internal/controller/clustermesh_controller.go +++ b/internal/controller/clustermesh_controller.go @@ -184,15 +184,20 @@ func (r *ClusterMeshReconciler) cleanupStaleSourceClusters(ctx context.Context, // Per-target deadline so one unreachable cluster cannot stall the // whole reconcile pass — see cleanupSweepTimeout for the rationale. - sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) - err := peer.DeleteStaleSourceClusters(sweepCtx, tgtClient, mesh.Name, sources) - cancel() - if err != nil { - log.Warn("cleaning stale source-cluster peers", - slog.String("target", name), - slog.String("error", err.Error()), - ) - } + // Anonymous function so defer cancel() fires per iteration instead + // of waiting for the enclosing reconcile to return. + func() { + sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) + defer cancel() + + err := peer.DeleteStaleSourceClusters(sweepCtx, tgtClient, mesh.Name, sources) + if err != nil { + log.Warn("cleaning stale source-cluster peers", + slog.String("target", name), + slog.String("error", err.Error()), + ) + } + }() } } @@ -229,10 +234,14 @@ func (r *ClusterMeshReconciler) cleanupOrphanMeshPeers(ctx context.Context, log } // Per-target deadline so one unreachable cluster cannot stall the - // whole reconcile pass — see cleanupSweepTimeout. - sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) - r.sweepOrphanPeersInCluster(sweepCtx, log, clusterName, tgtClient, living) - cancel() + // whole reconcile pass — see cleanupSweepTimeout. Wrapped in an + // anonymous function so defer cancel() fires per iteration. + func() { + sweepCtx, cancel := context.WithTimeout(ctx, cleanupSweepTimeout) + defer cancel() + + r.sweepOrphanPeersInCluster(sweepCtx, log, clusterName, tgtClient, living) + }() } // `namespace` is currently unused — see collectLivingMeshes for the