From dfa3a367d59be5be7759fa2468d816ffbd65f14b Mon Sep 17 00:00:00 2001 From: Caleb Xu Date: Wed, 16 Sep 2026 11:19:54 -0400 Subject: [PATCH] fix: prevent double-binding of ClusterLease to ClusterInstance Signed-off-by: Caleb Xu --- Makefile | 2 +- README.md | 4 + cmd/main.go | 9 +- .../controller/clusterinstance_reservation.go | 25 ++++ .../controller/clusterlease_controller.go | 83 ++++++++----- .../clusterlease_verification_test.go | 117 +++++++++++++++++- internal/controller/clusterpool_controller.go | 28 +++-- .../clusterpool_verification_test.go | 24 ++++ 8 files changed, 238 insertions(+), 54 deletions(-) create mode 100644 internal/controller/clusterinstance_reservation.go diff --git a/Makefile b/Makefile index 0dd263a..bc5ff0e 100644 --- a/Makefile +++ b/Makefile @@ -173,7 +173,7 @@ build: manifests generate fmt vet ## Build manager binary. .PHONY: run run: manifests generate fmt vet ## Run a controller from your host. - go run ./cmd/main.go + go run ./cmd/main.go --leader-elect=false # If you wish to build the manager image targeting other platforms you can use the --platform flag. # (i.e. docker build --platform linux/arm64). However, you must enable docker buildKit for it. diff --git a/README.md b/README.md index bfc4414..75e81f2 100644 --- a/README.md +++ b/README.md @@ -592,6 +592,10 @@ make docker-build docker-push IMG=/guestcluster-operator:tag make deploy IMG=/guestcluster-operator:tag ``` +Leader election is enabled by default because lease binding uses process-local +serialization. Use `--leader-elect=false` only when one manager process can +run against the cluster. + **Create pools** for the topologies you need. See `config/samples/` for ready-to-edit examples: `guestcluster_v1alpha1_clusterpool.yaml` for `crc`, and `guestcluster_v1alpha1_clusterpool_hcp.yaml` for `hcp`: diff --git a/cmd/main.go b/cmd/main.go index 6ec5755..e56065e 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -79,9 +79,9 @@ func main() { flag.StringVar(&metricsAddr, "metrics-bind-address", "0", "The address the metrics endpoint binds to. "+ "Use :8443 for HTTPS or :8080 for HTTP, or leave as 0 to disable the metrics service.") flag.StringVar(&probeAddr, "health-probe-bind-address", ":8081", "The address the probe endpoint binds to.") - flag.BoolVar(&enableLeaderElection, "leader-elect", false, + flag.BoolVar(&enableLeaderElection, "leader-elect", true, "Enable leader election for controller manager. "+ - "Enabling this will ensure there is only one active controller manager.") + "Keep this enabled unless only one manager process can run.") flag.BoolVar(&secureMetrics, "metrics-secure", true, "If set, the metrics endpoint is served securely via HTTPS. Use --metrics-secure=false to use HTTP instead.") flag.StringVar(&webhookCertPath, "webhook-cert-path", "", "The directory that contains the webhook certificate.") @@ -228,8 +228,9 @@ func main() { os.Exit(1) } if err := (&controller.ClusterLeaseReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + APIReader: mgr.GetAPIReader(), }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "ClusterLease") os.Exit(1) diff --git a/internal/controller/clusterinstance_reservation.go b/internal/controller/clusterinstance_reservation.go new file mode 100644 index 0000000..6407297 --- /dev/null +++ b/internal/controller/clusterinstance_reservation.go @@ -0,0 +1,25 @@ +/* +Copyright 2026. + +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 controller + +import "sync" + +// clusterInstanceReservation serializes every operation that can reserve or +// remove a ClusterInstance. The deployed configuration enables leader +// election so only one manager performs these operations at a time, while +// this mutex also serializes the lease and pool controllers inside it. +var clusterInstanceReservation sync.Mutex diff --git a/internal/controller/clusterlease_controller.go b/internal/controller/clusterlease_controller.go index 0d069ee..36e58f0 100644 --- a/internal/controller/clusterlease_controller.go +++ b/internal/controller/clusterlease_controller.go @@ -74,16 +74,10 @@ const ( // from a named ClusterPool, it finds a Ready ClusterInstance belonging to // that pool that no other lease currently claims, and binds it to this // lease with a SINGLE atomic write to the lease's own status (InstanceRef -// plus Phase=Bound). This is modeled directly on how the Kubernetes -// scheduler binds a Pod to a Node by writing Pod.Spec.NodeName: one -// authoritative pointer, on the demand object, written once. Unlike the -// scheduler analogy, this reconciler is the ONLY writer of that pointer, so -// there is no two-controllers-racing-to-bind concern. Critically, it never -// writes anything to the ClusterInstance side. This design eliminates the -// two-write partial-bind races, for example an instance claimed while its -// lease's own status lags behind, or a lease retried after a partial -// failure re-claiming a second instance, that motivated defensive patches -// in earlier iterations of this controller and ClusterPoolReconciler. +// plus Phase=Bound). Matching uses a live API read and is serialized with +// pool scale-down, so a stale informer cache cannot cause two leases to select +// or remove the same instance. The lease status remains the authoritative +// binding record, and this reconciler does not write ClusterInstance status. // // ClusterLeaseReconciler never creates new ClusterInstances itself; that // supply-side responsibility belongs entirely to ClusterPoolReconciler. On @@ -93,7 +87,8 @@ const ( // fresh replacement, the same as it does for any other capacity shortfall. type ClusterLeaseReconciler struct { client.Client - Scheme *runtime.Scheme + Scheme *runtime.Scheme + APIReader client.Reader } // +kubebuilder:rbac:groups=guestcluster.opdev.io,resources=clusterleases,verbs=get;list;watch;create;update;patch;delete @@ -106,7 +101,7 @@ func (r *ClusterLeaseReconciler) Reconcile(ctx context.Context, req ctrl.Request log := logf.FromContext(ctx) lease := &brokerv1alpha1.ClusterLease{} - if err := r.Get(ctx, req.NamespacedName, lease); err != nil { + if err := r.apiReader().Get(ctx, req.NamespacedName, lease); err != nil { if apierrors.IsNotFound(err) { log.V(1).Info("ClusterLease deleted, nothing to reconcile") return ctrl.Result{}, nil @@ -154,8 +149,32 @@ func (r *ClusterLeaseReconciler) Reconcile(ctx context.Context, req ctrl.Request func (r *ClusterLeaseReconciler) reconcilePending(ctx context.Context, lease *brokerv1alpha1.ClusterLease) (ctrl.Result, error) { log := logf.FromContext(ctx) + clusterInstanceReservation.Lock() + defer clusterInstanceReservation.Unlock() + + // Re-read the lease after acquiring the reservation. A second reconcile + // for the same lease can have started before the first one committed its + // status, and must not bind that lease to a second instance from a stale + // Pending object. + currentLease := &brokerv1alpha1.ClusterLease{} + if err := r.apiReader().Get(ctx, client.ObjectKeyFromObject(lease), currentLease); err != nil { + if apierrors.IsNotFound(err) { + return ctrl.Result{}, nil + } + return ctrl.Result{}, fmt.Errorf("refreshing ClusterLease before matching: %w", err) + } + if !currentLease.DeletionTimestamp.IsZero() || currentLease.Status.InstanceRef != nil { + return ctrl.Result{}, nil + } + switch currentLease.Status.Phase { + case "", brokerv1alpha1.PhaseLeasePending: + lease = currentLease + default: + return ctrl.Result{}, nil + } + instanceList := &brokerv1alpha1.ClusterInstanceList{} - if err := r.List(ctx, instanceList, + if err := r.apiReader().List(ctx, instanceList, client.InNamespace(lease.Namespace), client.MatchingLabels(resources.PoolLabels(lease.Spec.PoolRef.Name)), ); err != nil { @@ -171,15 +190,12 @@ func (r *ClusterLeaseReconciler) reconcilePending(ctx context.Context, lease *br // scale and avoids needing a field index for this one in-namespace, // same-pool lookup. siblingLeases := &brokerv1alpha1.ClusterLeaseList{} - if err := r.List(ctx, siblingLeases, client.InNamespace(lease.Namespace)); err != nil { + if err := r.apiReader().List(ctx, siblingLeases, client.InNamespace(lease.Namespace)); err != nil { return ctrl.Result{}, fmt.Errorf("listing sibling ClusterLeases: %w", err) } claimed := make(map[string]bool) for i := range siblingLeases.Items { l := &siblingLeases.Items[i] - if !l.DeletionTimestamp.IsZero() { - continue - } if l.Spec.PoolRef.Name != lease.Spec.PoolRef.Name { continue } @@ -224,31 +240,20 @@ func (r *ClusterLeaseReconciler) reconcilePending(ctx context.Context, lease *br return r.bind(ctx, lease, candidate) } -// bind copies the instance's kubeconfig Secret into a lease-owned Secret, -// and mirrors the instance's observed version, topology, and endpoint onto -// the lease status as the explicit CI outputs the acceptance criteria -// require. It then commits the binding itself via a SINGLE Status().Update -// on the lease (InstanceRef plus Phase=Bound), the one and only -// authoritative write of the lease-instance relationship anywhere in the -// operator. bind writes nothing to the ClusterInstance side. A -// resourceVersion conflict on this Status().Update, from another -// ClusterLease reconcile racing for the same instance, surfaces as an -// error, which controller-runtime retries against a freshly-Get'd lease on -// the next attempt. Because re-running reconcilePending re-derives the -// claimed set from scratch, two concurrent leases can never both claim the -// same instance, and a failed or retried attempt here can never leave a -// stray, unclaimed-by-any-lease write behind: if this call fails, it -// writes NOTHING. +// bind copies the instance's kubeconfig Secret into a lease-owned Secret and +// commits the binding via one Status().Update on the lease. The caller holds +// clusterInstanceReservation from matching through this update, so another +// lease or pool scale-down cannot act on the same instance concurrently. func (r *ClusterLeaseReconciler) bind(ctx context.Context, lease *brokerv1alpha1.ClusterLease, instance *brokerv1alpha1.ClusterInstance) (ctrl.Result, error) { srcSecret := &corev1.Secret{} srcKey := client.ObjectKey{Namespace: instance.Namespace, Name: instance.Status.KubeconfigSecretRef.Name} - if err := r.Get(ctx, srcKey, srcSecret); err != nil { + if err := r.apiReader().Get(ctx, srcKey, srcSecret); err != nil { return ctrl.Result{}, fmt.Errorf("fetching instance kubeconfig secret %s: %w", srcKey, err) } leaseSecretName := resources.LeaseKubeconfigSecretName(lease.Name) leaseSecret := &corev1.Secret{} - err := r.Get(ctx, client.ObjectKey{Namespace: lease.Namespace, Name: leaseSecretName}, leaseSecret) + err := r.apiReader().Get(ctx, client.ObjectKey{Namespace: lease.Namespace, Name: leaseSecretName}, leaseSecret) switch { case apierrors.IsNotFound(err): leaseSecret = &corev1.Secret{ @@ -346,6 +351,9 @@ func (r *ClusterLeaseReconciler) reconcileDelete(ctx context.Context, lease *bro return ctrl.Result{}, nil } + clusterInstanceReservation.Lock() + defer clusterInstanceReservation.Unlock() + if err := r.releaseBoundInstance(ctx, lease); err != nil { return ctrl.Result{}, err } @@ -396,6 +404,13 @@ func (r *ClusterLeaseReconciler) setPendingCondition(ctx context.Context, lease return nil } +func (r *ClusterLeaseReconciler) apiReader() client.Reader { + if r.APIReader != nil { + return r.APIReader + } + return r.Client +} + // leasesForInstance maps a ClusterInstance event to reconcile.Requests for // every non-terminal ClusterLease of the same pool that has not yet // claimed an instance (that is, still Pending). This lets a newly-Ready, or diff --git a/internal/controller/clusterlease_verification_test.go b/internal/controller/clusterlease_verification_test.go index 03f2e13..0f22f2d 100644 --- a/internal/controller/clusterlease_verification_test.go +++ b/internal/controller/clusterlease_verification_test.go @@ -37,6 +37,8 @@ package controller import ( "context" "fmt" + "sync" + "sync/atomic" "time" . "github.com/onsi/ginkgo/v2" @@ -65,6 +67,46 @@ func (c *failOnceDeleteClient) Delete(ctx context.Context, obj client.Object, op return c.Client.Delete(ctx, obj, opts...) } +type delayedCacheClient struct { + client.Client + leases *brokerv1alpha1.ClusterLeaseList + instances *brokerv1alpha1.ClusterInstanceList +} + +type blockingInstanceListReader struct { + client.Reader + firstEntered chan struct{} + releaseFirst chan struct{} + calls atomic.Int32 +} + +func (r *blockingInstanceListReader) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + if _, ok := list.(*brokerv1alpha1.ClusterInstanceList); ok { + if r.calls.Add(1) == 1 { + close(r.firstEntered) + select { + case <-r.releaseFirst: + case <-ctx.Done(): + return ctx.Err() + } + } + } + return r.Reader.List(ctx, list, opts...) +} + +func (c *delayedCacheClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error { + switch typedList := list.(type) { + case *brokerv1alpha1.ClusterLeaseList: + *typedList = *c.leases.DeepCopy() + return nil + case *brokerv1alpha1.ClusterInstanceList: + *typedList = *c.instances.DeepCopy() + return nil + default: + return c.Client.List(ctx, list, opts...) + } +} + var _ = func() bool { for _, topology := range allVerificationTopologies { registerClusterLeaseVerificationSpecs(topology) @@ -135,12 +177,14 @@ func registerClusterLeaseVerificationSpecs(topology brokerv1alpha1.ClusterTopolo return lease } - reconcileLease := func(lease *brokerv1alpha1.ClusterLease) reconcile.Result { - r := &ClusterLeaseReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()} + reconcileLeaseWith := func(r *ClusterLeaseReconciler, lease *brokerv1alpha1.ClusterLease) reconcile.Result { res, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(lease)}) Expect(err).NotTo(HaveOccurred()) return res } + reconcileLease := func(lease *brokerv1alpha1.ClusterLease) reconcile.Result { + return reconcileLeaseWith(&ClusterLeaseReconciler{Client: k8sClient, Scheme: k8sClient.Scheme()}, lease) + } AfterEach(func() { Expect(client.IgnoreNotFound(k8sClient.DeleteAllOf(ctx, &brokerv1alpha1.ClusterLease{}, client.InNamespace(namespace)))).To(Succeed()) @@ -203,6 +247,75 @@ func registerClusterLeaseVerificationSpecs(topology brokerv1alpha1.ClusterTopolo _ = finalInst // instance-side state is a derived projection, not asserted here }) + It("serializes concurrent binding when the lease cache is delayed", func() { + newReadyInstance("verify-delayed-cache-inst") + first := newPendingLease("verify-delayed-cache-first", nil) + second := newPendingLease("verify-delayed-cache-second", nil) + + staleLeases := &brokerv1alpha1.ClusterLeaseList{} + Expect(k8sClient.List(ctx, staleLeases, client.InNamespace(namespace))).To(Succeed()) + staleInstances := &brokerv1alpha1.ClusterInstanceList{} + Expect(k8sClient.List(ctx, staleInstances, client.InNamespace(namespace))).To(Succeed()) + staleClient := &delayedCacheClient{ + Client: k8sClient, + leases: staleLeases, + instances: staleInstances, + } + liveReader := &blockingInstanceListReader{ + Reader: k8sClient, + firstEntered: make(chan struct{}), + releaseFirst: make(chan struct{}), + } + releaseFirst := sync.OnceFunc(func() { close(liveReader.releaseFirst) }) + DeferCleanup(releaseFirst) + r := &ClusterLeaseReconciler{ + Client: staleClient, + Scheme: k8sClient.Scheme(), + APIReader: liveReader, + } + + errs := make(chan error, 2) + reconcileAsync := func(lease *brokerv1alpha1.ClusterLease) { + _, err := r.Reconcile(ctx, reconcile.Request{NamespacedName: client.ObjectKeyFromObject(lease)}) + errs <- err + } + + go reconcileAsync(first) + Eventually(liveReader.firstEntered).Should(BeClosed()) + go reconcileAsync(second) + Consistently(liveReader.calls.Load, 100*time.Millisecond, 10*time.Millisecond).Should(Equal(int32(1)), + "a second matcher must not enter the reservation section concurrently") + releaseFirst() + Eventually(errs).Should(Receive(BeNil())) + Eventually(errs).Should(Receive(BeNil())) + + boundFirst := &brokerv1alpha1.ClusterLease{} + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(first), boundFirst)).To(Succeed()) + Expect(boundFirst.Status.Phase).To(Equal(brokerv1alpha1.PhaseLeaseBound)) + stillPending := &brokerv1alpha1.ClusterLease{} + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(second), stillPending)).To(Succeed()) + Expect(stillPending.Status.Phase).To(Equal(brokerv1alpha1.PhaseLeasePending)) + Expect(stillPending.Status.InstanceRef).To(BeNil()) + }) + + It("keeps a deleting lease's instance claimed until cleanup starts", func() { + inst := newReadyInstance("verify-deleting-claim-inst") + claimant := newPendingLease("verify-deleting-claim-first", nil) + claimant.Status.Phase = brokerv1alpha1.PhaseLeaseBound + claimant.Status.InstanceRef = &corev1.LocalObjectReference{Name: inst.Name} + Expect(k8sClient.Status().Update(ctx, claimant)).To(Succeed()) + Expect(k8sClient.Delete(ctx, claimant)).To(Succeed()) + + second := newPendingLease("verify-deleting-claim-second", nil) + reconcileLease(second) + + stillPending := &brokerv1alpha1.ClusterLease{} + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(second), stillPending)).To(Succeed()) + Expect(stillPending.Status.Phase).To(Equal(brokerv1alpha1.PhaseLeasePending)) + Expect(stillPending.Status.InstanceRef).To(BeNil()) + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(inst), &brokerv1alpha1.ClusterInstance{})).To(Succeed()) + }) + It("releases on TTL expiry: deletes both the lease and its claimed instance", func() { inst := newReadyInstance("verify-ttl-inst") lease := newPendingLease("verify-ttl-lease", &metav1.Duration{Duration: time.Minute}) diff --git a/internal/controller/clusterpool_controller.go b/internal/controller/clusterpool_controller.go index 416a769..7e65bf2 100644 --- a/internal/controller/clusterpool_controller.go +++ b/internal/controller/clusterpool_controller.go @@ -98,12 +98,10 @@ const ( // --scale-down-unneeded-time: a Ready, unclaimed instance must stay idle // for at least this long before it becomes eligible for trimming as // excess. This is defense-in-depth against thrashing a spare that is - // about to be claimed. The single-write, assume-immediately binding - // model (see clusterlease_controller.go) already closes most such races - // structurally, but a short grace period costs nothing and protects - // against races not yet anticipated (for example, an unrelated - // conflict/backoff that delays a lease's own reconcile between finding - // a candidate and committing the bind). + // about to be claimed. The live-read and reservation model (see + // clusterlease_controller.go) closes the controller race, but a short + // grace period costs nothing and protects against unrelated delays between + // finding a candidate and committing the bind. scaleDownStabilityPeriod = 2 * time.Minute ) @@ -139,6 +137,9 @@ func (r *ClusterPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) } } + clusterInstanceReservation.Lock() + defer clusterInstanceReservation.Unlock() + instanceList := &brokerv1alpha1.ClusterInstanceList{} if err := r.APIReader.List(ctx, instanceList, client.InNamespace(pool.Namespace), @@ -247,10 +248,11 @@ func (r *ClusterPoolReconciler) Reconcile(ctx context.Context, req ctrl.Request) // which ClusterInstances are claimed and how much unclaimed demand // (pendingDemand) targets pool poolName. // -// claimed records, by ClusterInstance name, every instance named by some -// non-deleted ClusterLease's Status.InstanceRef, the single source of truth -// for the lease-instance binding (see clusterlease_types.go). This is the -// ONLY place that determines "is this instance in use"; +// claimed records, by ClusterInstance name, every instance named by a +// ClusterLease's Status.InstanceRef, including a deleting lease until its +// instance is unavailable. Status.InstanceRef is the single source of truth +// for the lease-instance binding (see clusterlease_types.go). This is the ONLY +// place that determines "is this instance in use"; // ClusterInstance.Status.LeaseRef is a read-only derived projection and // Reconcile never consults it for accounting decisions. // ClusterLeaseReconciler commits a claim with a single atomic write to the @@ -273,12 +275,12 @@ func computeLeaseAccounting(leaseList *brokerv1alpha1.ClusterLeaseList, poolName claimed = make(map[string]bool) for i := range leaseList.Items { l := &leaseList.Items[i] - if !l.DeletionTimestamp.IsZero() { - continue - } if l.Status.InstanceRef != nil { claimed[l.Status.InstanceRef.Name] = true } + if !l.DeletionTimestamp.IsZero() { + continue + } if l.Spec.PoolRef.Name != poolName { continue } diff --git a/internal/controller/clusterpool_verification_test.go b/internal/controller/clusterpool_verification_test.go index e9508be..71b6487 100644 --- a/internal/controller/clusterpool_verification_test.go +++ b/internal/controller/clusterpool_verification_test.go @@ -246,6 +246,30 @@ func registerClusterPoolVerificationSpecs(topology brokerv1alpha1.ClusterTopolog "a claimed instance must never be scaled down even though minSize=warmSpares=0") }) + It("does not scale down an instance claimed by a deleting ClusterLease", func() { + pool := newVerifyPool("pool-neverdelete-deleting", 4, 0, 0) + Expect(k8sClient.Create(ctx, pool)).To(Succeed()) + inst := newVerifyInstance("inst-deleting-claimed", pool.Name, brokerv1alpha1.PhaseReady) + lease := newVerifyLease("lease-deleting-claims-it", pool.Name, inst.Name) + lease.Finalizers = []string{leaseFinalizer} + Expect(k8sClient.Update(ctx, lease)).To(Succeed()) + Expect(k8sClient.Delete(ctx, lease)).To(Succeed()) + + fresh := &brokerv1alpha1.ClusterInstance{} + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(inst), fresh)).To(Succeed()) + for i := range fresh.Status.Conditions { + if fresh.Status.Conditions[i].Type == conditionTypeReady { + fresh.Status.Conditions[i].LastTransitionTime = metav1.NewTime(time.Now().Add(-2 * scaleDownStabilityPeriod)) + } + } + Expect(k8sClient.Status().Update(ctx, fresh)).To(Succeed()) + + reconcilePool(pool) + + Expect(k8sClient.Get(ctx, client.ObjectKeyFromObject(inst), &brokerv1alpha1.ClusterInstance{})).To(Succeed(), + "a deleting lease must continue to protect its instance until lease cleanup deletes it") + }) + It("respects the scale-down stability window: does not delete a freshly-Ready excess instance, but does once it has been idle long enough", func() { pool := newVerifyPool("pool-stability", 4, 0, 0) Expect(k8sClient.Create(ctx, pool)).To(Succeed())