From 5ffb6d4319874a5e7a6d8b3611fc922148e0bd23 Mon Sep 17 00:00:00 2001 From: zgjia Date: Wed, 3 Jun 2026 17:24:03 +0800 Subject: [PATCH] fix: stop reconciling business clusters in standby Make standby handling cluster-aware so standby management clusters continue reconciling the global Cluster while skipping non-global business cluster resources. Use global-aware wrappers for Cluster, Machine, MachineSet, MachineDeployment, MachineHealthCheck, MachinePool, topology, and ClusterResourceSetBinding reconcilers. Leave global/no-cluster controllers such as ClusterClass, ExtensionConfig, and CRD migrator unguarded for now. Filter ClusterResourceSet target clusters in standby so only global clusters are applied. During deletion, keep the ClusterResourceSet finalizer and requeue if standby filtering skipped non-global clusters, so their ClusterResourceSetBindings can be cleaned up after failback. Use GlobalClusterName for the reserved global cluster name, and do not add standby blocking at the ClusterCache layer. Guard the Machine infrastructure providerID fallback against nil values. --- .../controllers/machinepool_controller.go | 9 +- .../controllers/cluster/cluster_controller.go | 3 +- .../clusterresourceset_controller.go | 61 ++++- .../clusterresourceset_controller_test.go | 87 +++++++ .../clusterresourcesetbinding_controller.go | 15 +- .../controllers/machine/machine_controller.go | 11 +- .../machinedeployment_controller.go | 9 +- .../machinehealthcheck_controller.go | 9 +- .../machineset/machineset_controller.go | 9 +- .../topology/cluster/cluster_controller.go | 3 +- .../machinedeployment_controller.go | 9 +- .../machineset/machineset_controller.go | 9 +- main.go | 4 + pkg/standby/standby.go | 191 ++++++++++++++++ pkg/standby/standby_test.go | 216 ++++++++++++++++++ 15 files changed, 627 insertions(+), 18 deletions(-) create mode 100644 pkg/standby/standby.go create mode 100644 pkg/standby/standby_test.go diff --git a/exp/internal/controllers/machinepool_controller.go b/exp/internal/controllers/machinepool_controller.go index b3ab1c1f03b8..2b66a1e21336 100644 --- a/exp/internal/controllers/machinepool_controller.go +++ b/exp/internal/controllers/machinepool_controller.go @@ -46,6 +46,7 @@ import ( "sigs.k8s.io/cluster-api/controllers/external" expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1" "sigs.k8s.io/cluster-api/internal/util/ssa" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/conditions" "sigs.k8s.io/cluster-api/util/finalizers" @@ -132,7 +133,13 @@ func (r *MachinePoolReconciler) SetupWithManager(ctx context.Context, mgr ctrl.M ), ). WatchesRawSource(r.ClusterCache.GetClusterSource("machinepool", clusterToMachinePools)). - Build(r) + Build(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machinepool", + func() client.Object { return &expv1.MachinePool{} }, + func(obj client.Object) string { return obj.(*expv1.MachinePool).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/cluster/cluster_controller.go b/internal/controllers/cluster/cluster_controller.go index 829cad4ce22e..682835f00016 100644 --- a/internal/controllers/cluster/cluster_controller.go +++ b/internal/controllers/cluster/cluster_controller.go @@ -49,6 +49,7 @@ import ( expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1" "sigs.k8s.io/cluster-api/feature" "sigs.k8s.io/cluster-api/internal/hooks" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/collections" "sigs.k8s.io/cluster-api/util/conditions" @@ -122,7 +123,7 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt c, err := b. WithOptions(options). WithEventFilter(predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue)). - Build(r) + Build(standby.WrapClusterReconcilerWithConfigMapDetector(mgr.GetAPIReader(), "cluster", r)) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") diff --git a/internal/controllers/clusterresourceset/clusterresourceset_controller.go b/internal/controllers/clusterresourceset/clusterresourceset_controller.go index fb11c980ed7f..cd933bd17316 100644 --- a/internal/controllers/clusterresourceset/clusterresourceset_controller.go +++ b/internal/controllers/clusterresourceset/clusterresourceset_controller.go @@ -44,6 +44,7 @@ import ( clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" "sigs.k8s.io/cluster-api/controllers/clustercache" resourcepredicates "sigs.k8s.io/cluster-api/internal/controllers/clusterresourceset/predicates" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/conditions" v1beta2conditions "sigs.k8s.io/cluster-api/util/conditions/v1beta2" @@ -66,6 +67,9 @@ type Reconciler struct { Client client.Client ClusterCache clustercache.ClusterCache + // StandbyDetector detects if the management cluster is in standby mode. + StandbyDetector standby.Detector + // WatchFilterValue is the label value used to filter events prior to reconciliation. WatchFilterValue string } @@ -74,6 +78,9 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt if r.Client == nil || r.ClusterCache == nil { return errors.New("Client and ClusterCache must not be nil") } + if r.StandbyDetector == nil { + r.StandbyDetector = standby.NewConfigMapDetector(mgr.GetAPIReader()) + } predicateLog := ctrl.LoggerFrom(ctx).WithValues("controller", "clusterresourceset") err := ctrl.NewControllerManagedBy(mgr). @@ -181,10 +188,14 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Re }) return ctrl.Result{}, err } + clusters, skippedNonGlobalClusters, err := r.filterClustersForStandby(ctx, clusters) + if err != nil { + return ctrl.Result{}, err + } // Handle deletion reconciliation loop. if !clusterResourceSet.ObjectMeta.DeletionTimestamp.IsZero() { - return ctrl.Result{}, r.reconcileDelete(ctx, clusters, clusterResourceSet) + return r.reconcileDelete(ctx, clusters, clusterResourceSet, skippedNonGlobalClusters) } errs := []error{} @@ -219,7 +230,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req ctrl.Request) (_ ctrl.Re } // reconcileDelete removes the deleted ClusterResourceSet from all the ClusterResourceSetBindings it is added to. -func (r *Reconciler) reconcileDelete(ctx context.Context, clusters []*clusterv1.Cluster, crs *addonsv1.ClusterResourceSet) error { +func (r *Reconciler) reconcileDelete(ctx context.Context, clusters []*clusterv1.Cluster, crs *addonsv1.ClusterResourceSet, skippedNonGlobalClusters bool) (ctrl.Result, error) { for _, cluster := range clusters { log := ctrl.LoggerFrom(ctx, "Cluster", klog.KObj(cluster)) @@ -230,16 +241,20 @@ func (r *Reconciler) reconcileDelete(ctx context.Context, clusters []*clusterv1. } if err := r.Client.Get(ctx, clusterResourceSetBindingKey, clusterResourceSetBinding); err != nil { if !apierrors.IsNotFound(err) { - return errors.Wrapf(err, "failed to get ClusterResourceSetBinding during ClusterResourceSet deletion") + return ctrl.Result{}, errors.Wrapf(err, "failed to get ClusterResourceSetBinding during ClusterResourceSet deletion") + } + if skippedNonGlobalClusters { + logSkipRemoveFinalizerForStandby(ctx, crs) + return ctrl.Result{RequeueAfter: standby.RequeueAfter}, nil } controllerutil.RemoveFinalizer(crs, addonsv1.ClusterResourceSetFinalizer) - return nil + return ctrl.Result{}, nil } // Initialize the patch helper. patchHelper, err := patch.NewHelper(clusterResourceSetBinding, r.Client) if err != nil { - return err + return ctrl.Result{}, err } clusterResourceSetBinding.RemoveBinding(crs) @@ -256,12 +271,20 @@ func (r *Reconciler) reconcileDelete(ctx context.Context, clusters []*clusterv1. log.Error(err, "Failed to delete empty ClusterResourceSetBinding") } } else if err := patchHelper.Patch(ctx, clusterResourceSetBinding); err != nil { - return err + return ctrl.Result{}, err } } + if skippedNonGlobalClusters { + logSkipRemoveFinalizerForStandby(ctx, crs) + return ctrl.Result{RequeueAfter: standby.RequeueAfter}, nil + } controllerutil.RemoveFinalizer(crs, addonsv1.ClusterResourceSetFinalizer) - return nil + return ctrl.Result{}, nil +} + +func logSkipRemoveFinalizerForStandby(ctx context.Context, crs *addonsv1.ClusterResourceSet) { + ctrl.LoggerFrom(ctx).Info("Refusing to remove ClusterResourceSet finalizer because management cluster is DR standby and non-global clusters were skipped", "ClusterResourceSet", klog.KObj(crs), "finalizer", addonsv1.ClusterResourceSetFinalizer, "globalCluster", standby.GlobalClusterName, "requeueAfter", standby.RequeueAfter) } // getClustersByClusterResourceSetSelector fetches Clusters matched by the ClusterResourceSet's label selector that are in the same namespace as the ClusterResourceSet object. @@ -294,6 +317,30 @@ func (r *Reconciler) getClustersByClusterResourceSetSelector(ctx context.Context return clusters, nil } +func (r *Reconciler) filterClustersForStandby(ctx context.Context, clusters []*clusterv1.Cluster) ([]*clusterv1.Cluster, bool, error) { + if r.StandbyDetector == nil { + return clusters, false, nil + } + isStandby, err := r.StandbyDetector.IsStandby(ctx) + if err != nil { + return nil, false, err + } + if !isStandby { + return clusters, false, nil + } + + filtered := []*clusterv1.Cluster{} + skippedNonGlobalClusters := false + for _, cluster := range clusters { + if cluster.Name == standby.GlobalClusterName { + filtered = append(filtered, cluster) + continue + } + skippedNonGlobalClusters = true + } + return filtered, skippedNonGlobalClusters, nil +} + // ApplyClusterResourceSet applies resources in a ClusterResourceSet to a Cluster. Once applied, a record will be added to the // cluster's ClusterResourceSetBinding. // In ApplyOnce strategy, resources are applied only once to a particular cluster. ClusterResourceSetBinding is used to check if a resource is applied before. diff --git a/internal/controllers/clusterresourceset/clusterresourceset_controller_test.go b/internal/controllers/clusterresourceset/clusterresourceset_controller_test.go index 231b9e89682a..40decb7d5965 100644 --- a/internal/controllers/clusterresourceset/clusterresourceset_controller_test.go +++ b/internal/controllers/clusterresourceset/clusterresourceset_controller_test.go @@ -17,6 +17,7 @@ limitations under the License. package clusterresourceset import ( + "context" "crypto/sha1" //nolint: gosec "fmt" "testing" @@ -34,6 +35,7 @@ import ( addonsv1 "sigs.k8s.io/cluster-api/api/addons/v1beta1" clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" "sigs.k8s.io/cluster-api/internal/test/envtest" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/conditions" v1beta2conditions "sigs.k8s.io/cluster-api/util/conditions/v1beta2" @@ -43,6 +45,91 @@ const ( timeout = time.Second * 15 ) +type crsStandbyDetector struct { + standby bool + err error +} + +func (d crsStandbyDetector) IsStandby(context.Context) (bool, error) { + return d.standby, d.err +} + +func TestFilterClustersForStandby(t *testing.T) { + testCases := []struct { + name string + detector standby.Detector + clusters []*clusterv1.Cluster + expectedNames []string + expectedSkipped bool + expectError bool + }{ + { + name: "active returns all clusters", + detector: crsStandbyDetector{}, + clusters: []*clusterv1.Cluster{{ObjectMeta: metav1.ObjectMeta{Name: standby.GlobalClusterName}}, {ObjectMeta: metav1.ObjectMeta{Name: "business"}}}, + expectedNames: []string{standby.GlobalClusterName, "business"}, + }, + { + name: "standby returns global cluster only", + detector: crsStandbyDetector{standby: true}, + clusters: []*clusterv1.Cluster{{ObjectMeta: metav1.ObjectMeta{Name: standby.GlobalClusterName}}, {ObjectMeta: metav1.ObjectMeta{Name: "business"}}}, + expectedNames: []string{standby.GlobalClusterName}, + expectedSkipped: true, + }, + { + name: "standby returns empty without global cluster", + detector: crsStandbyDetector{standby: true}, + clusters: []*clusterv1.Cluster{{ObjectMeta: metav1.ObjectMeta{Name: "business"}}}, + expectedNames: []string{}, + expectedSkipped: true, + }, + { + name: "detector error fails closed", + detector: crsStandbyDetector{err: errors.New("detector error")}, + clusters: []*clusterv1.Cluster{{ObjectMeta: metav1.ObjectMeta{Name: standby.GlobalClusterName}}}, + expectError: true, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + g := NewWithT(t) + r := &Reconciler{StandbyDetector: tc.detector} + + clusters, skipped, err := r.filterClustersForStandby(context.Background(), tc.clusters) + if tc.expectError { + g.Expect(err).To(HaveOccurred()) + return + } + g.Expect(err).NotTo(HaveOccurred()) + + clusterNames := []string{} + for _, cluster := range clusters { + clusterNames = append(clusterNames, cluster.Name) + } + g.Expect(clusterNames).To(Equal(tc.expectedNames)) + g.Expect(skipped).To(Equal(tc.expectedSkipped)) + }) + } +} + +func TestReconcileDeleteSkipsFinalizerRemovalForStandbySkippedClusters(t *testing.T) { + g := NewWithT(t) + clusterResourceSet := &addonsv1.ClusterResourceSet{ + ObjectMeta: metav1.ObjectMeta{ + Name: "crs", + Namespace: metav1.NamespaceDefault, + Finalizers: []string{addonsv1.ClusterResourceSetFinalizer}, + }, + } + r := &Reconciler{} + + res, err := r.reconcileDelete(context.Background(), nil, clusterResourceSet, true) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res.RequeueAfter).To(Equal(standby.RequeueAfter)) + g.Expect(clusterResourceSet.Finalizers).To(ContainElement(addonsv1.ClusterResourceSetFinalizer)) +} + func TestClusterResourceSetReconciler(t *testing.T) { var ( clusterResourceSetName string diff --git a/internal/controllers/clusterresourcesetbinding/clusterresourcesetbinding_controller.go b/internal/controllers/clusterresourcesetbinding/clusterresourcesetbinding_controller.go index 8ed1601acd3a..9f874a16c5b5 100644 --- a/internal/controllers/clusterresourcesetbinding/clusterresourcesetbinding_controller.go +++ b/internal/controllers/clusterresourcesetbinding/clusterresourcesetbinding_controller.go @@ -34,6 +34,7 @@ import ( clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" "sigs.k8s.io/cluster-api/feature" "sigs.k8s.io/cluster-api/internal/hooks" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/patch" "sigs.k8s.io/cluster-api/util/predicates" @@ -64,7 +65,19 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ). WithOptions(options). WithEventFilter(predicates.ResourceNotPausedAndHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue)). - Complete(r) + Complete(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "clusterresourcesetbinding", + func() client.Object { return &addonsv1.ClusterResourceSetBinding{} }, + func(obj client.Object) string { + binding := obj.(*addonsv1.ClusterResourceSetBinding) + if binding.Spec.ClusterName != "" { + return binding.Spec.ClusterName + } + return binding.Name + }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/machine/machine_controller.go b/internal/controllers/machine/machine_controller.go index 0488591479a0..8ca7a731dac3 100644 --- a/internal/controllers/machine/machine_controller.go +++ b/internal/controllers/machine/machine_controller.go @@ -55,6 +55,7 @@ import ( "sigs.k8s.io/cluster-api/feature" "sigs.k8s.io/cluster-api/internal/contract" "sigs.k8s.io/cluster-api/internal/controllers/machine/drain" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/cache" @@ -172,7 +173,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt handler.EnqueueRequestsFromMapFunc(mdToMachines), builder.WithPredicates(predicates.ResourceIsChanged(mgr.GetScheme(), *r.predicateLog)), ). - Build(r) + Build(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machine", + func() client.Object { return &clusterv1.Machine{} }, + func(obj client.Object) string { return obj.(*clusterv1.Machine).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } @@ -726,7 +733,7 @@ func (r *Reconciler) isDeleteNodeAllowed(ctx context.Context, cluster *clusterv1 providerID = *machine.Spec.ProviderID } else if infraMachine != nil { // Fallback to retrieve from infraMachine. - if providerIDFromInfraMachine, err := contract.InfrastructureMachine().ProviderID().Get(infraMachine); err == nil { + if providerIDFromInfraMachine, err := contract.InfrastructureMachine().ProviderID().Get(infraMachine); err == nil && providerIDFromInfraMachine != nil { providerID = *providerIDFromInfraMachine } } diff --git a/internal/controllers/machinedeployment/machinedeployment_controller.go b/internal/controllers/machinedeployment/machinedeployment_controller.go index cd0194e52026..80fdcc844eeb 100644 --- a/internal/controllers/machinedeployment/machinedeployment_controller.go +++ b/internal/controllers/machinedeployment/machinedeployment_controller.go @@ -39,6 +39,7 @@ import ( clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" "sigs.k8s.io/cluster-api/controllers/external" "sigs.k8s.io/cluster-api/internal/util/ssa" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/conditions" utilconversion "sigs.k8s.io/cluster-api/util/conversion" @@ -108,7 +109,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt predicates.ClusterPausedTransitions(mgr.GetScheme(), predicateLog), )), // TODO: should this wait for Cluster.Status.InfrastructureReady similar to Infra Machine resources? - ).Complete(r) + ).Complete(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machinedeployment", + func() client.Object { return &clusterv1.MachineDeployment{} }, + func(obj client.Object) string { return obj.(*clusterv1.MachineDeployment).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go b/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go index 536da8c61762..71270b0597a9 100644 --- a/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go +++ b/internal/controllers/machinehealthcheck/machinehealthcheck_controller.go @@ -49,6 +49,7 @@ import ( "sigs.k8s.io/cluster-api/controllers/clustercache" "sigs.k8s.io/cluster-api/controllers/external" "sigs.k8s.io/cluster-api/internal/controllers/machine" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/conditions" @@ -118,7 +119,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ), ). WatchesRawSource(r.ClusterCache.GetClusterSource("machinehealthcheck", r.clusterToMachineHealthCheck)). - Build(r) + Build(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machinehealthcheck", + func() client.Object { return &clusterv1.MachineHealthCheck{} }, + func(obj client.Object) string { return obj.(*clusterv1.MachineHealthCheck).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/machineset/machineset_controller.go b/internal/controllers/machineset/machineset_controller.go index e64fc8f41f4f..0bffbe958c95 100644 --- a/internal/controllers/machineset/machineset_controller.go +++ b/internal/controllers/machineset/machineset_controller.go @@ -53,6 +53,7 @@ import ( "sigs.k8s.io/cluster-api/internal/controllers/machinedeployment/mdutil" topologynames "sigs.k8s.io/cluster-api/internal/topology/names" "sigs.k8s.io/cluster-api/internal/util/ssa" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/collections" @@ -148,7 +149,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ), ). WatchesRawSource(r.ClusterCache.GetClusterSource("machineset", clusterToMachineSets)). - Complete(r) + Complete(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machineset", + func() client.Object { return &clusterv1.MachineSet{} }, + func(obj client.Object) string { return obj.(*clusterv1.MachineSet).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/topology/cluster/cluster_controller.go b/internal/controllers/topology/cluster/cluster_controller.go index 6100a4f3c7f2..3f016bdb855c 100644 --- a/internal/controllers/topology/cluster/cluster_controller.go +++ b/internal/controllers/topology/cluster/cluster_controller.go @@ -59,6 +59,7 @@ import ( "sigs.k8s.io/cluster-api/internal/hooks" "sigs.k8s.io/cluster-api/internal/util/ssa" "sigs.k8s.io/cluster-api/internal/webhooks" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/conditions" @@ -145,7 +146,7 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ). WithOptions(options). WithEventFilter(predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue)). - Build(r) + Build(standby.WrapClusterReconcilerWithConfigMapDetector(mgr.GetAPIReader(), "clustertopology", r)) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") diff --git a/internal/controllers/topology/machinedeployment/machinedeployment_controller.go b/internal/controllers/topology/machinedeployment/machinedeployment_controller.go index 4c355efc9faf..3551c751c0da 100644 --- a/internal/controllers/topology/machinedeployment/machinedeployment_controller.go +++ b/internal/controllers/topology/machinedeployment/machinedeployment_controller.go @@ -33,6 +33,7 @@ import ( clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" "sigs.k8s.io/cluster-api/internal/controllers/topology/machineset" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/finalizers" @@ -91,7 +92,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ), ), ). - Complete(r) + Complete(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machinedeploymenttopology", + func() client.Object { return &clusterv1.MachineDeployment{} }, + func(obj client.Object) string { return obj.(*clusterv1.MachineDeployment).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/internal/controllers/topology/machineset/machineset_controller.go b/internal/controllers/topology/machineset/machineset_controller.go index aa2b52132aac..6aced6c7c42b 100644 --- a/internal/controllers/topology/machineset/machineset_controller.go +++ b/internal/controllers/topology/machineset/machineset_controller.go @@ -34,6 +34,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/handler" clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util" "sigs.k8s.io/cluster-api/util/annotations" "sigs.k8s.io/cluster-api/util/finalizers" @@ -93,7 +94,13 @@ func (r *Reconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, opt ), ), ). - Complete(r) + Complete(standby.WrapClusterNamedReconcilerWithConfigMapDetector( + mgr.GetAPIReader(), + "machinesettopology", + func() client.Object { return &clusterv1.MachineSet{} }, + func(obj client.Object) string { return obj.(*clusterv1.MachineSet).Spec.ClusterName }, + r, + )) if err != nil { return errors.Wrap(err, "failed setting up with a controller manager") } diff --git a/main.go b/main.go index 5d980758f469..80e92eb82f5f 100644 --- a/main.go +++ b/main.go @@ -80,6 +80,7 @@ import ( internalruntimeclient "sigs.k8s.io/cluster-api/internal/runtime/client" runtimeregistry "sigs.k8s.io/cluster-api/internal/runtime/registry" runtimewebhooks "sigs.k8s.io/cluster-api/internal/webhooks/runtime" + "sigs.k8s.io/cluster-api/pkg/standby" "sigs.k8s.io/cluster-api/util/apiwarnings" "sigs.k8s.io/cluster-api/util/flags" "sigs.k8s.io/cluster-api/version" @@ -182,6 +183,9 @@ func InitFlags(fs *pflag.FlagSet) { fs.StringVar(&watchFilterValue, "watch-filter", "", fmt.Sprintf("Label value that the controller watches to reconcile cluster-api objects. Label key is always %s. If unspecified, the controller watches for all cluster-api objects.", clusterv1.WatchLabel)) + fs.StringVar(&standby.ConfigMapNamespace, "system-namespace", standby.DefaultConfigMapNamespace, + "Namespace containing system resources used by the controller manager.") + fs.StringVar(&profilerAddress, "profiler-address", "", "Bind address to expose the pprof profiler (e.g. localhost:6060)") diff --git a/pkg/standby/standby.go b/pkg/standby/standby.go new file mode 100644 index 000000000000..9507bdf48e8e --- /dev/null +++ b/pkg/standby/standby.go @@ -0,0 +1,191 @@ +/* +Copyright 2026 The Kubernetes 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 standby + +import ( + "context" + "time" + + "github.com/pkg/errors" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/types" + "k8s.io/klog/v2" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +const ( + DefaultConfigMapNamespace = "cpaas-system" + ConfigMapName = "etcd-sync" + DefaultRequeueAfter = 30 * time.Second + GlobalClusterName = "global" +) + +var ( + ConfigMapNamespace = DefaultConfigMapNamespace + RequeueAfter = DefaultRequeueAfter +) + +type Detector interface { + IsStandby(ctx context.Context) (bool, error) +} + +type ConfigMapDetector struct { + Client client.Reader +} + +type ClusterNameGetter func(client.Object) string + +type guardedReconciler struct { + detector Detector + controllerName string + inner reconcile.Reconciler +} + +type guardedClusterReconciler struct { + detector Detector + controllerName string + inner reconcile.Reconciler +} + +type guardedClusterNamedReconciler struct { + detector Detector + reader client.Reader + controllerName string + newObject func() client.Object + getClusterName ClusterNameGetter + inner reconcile.Reconciler +} + +func NewConfigMapDetector(reader client.Reader) *ConfigMapDetector { + return &ConfigMapDetector{Client: reader} +} + +func (d *ConfigMapDetector) IsStandby(ctx context.Context) (bool, error) { + configMap := &corev1.ConfigMap{} + configMapKey := types.NamespacedName{Namespace: ConfigMapNamespace, Name: ConfigMapName} + if err := d.Client.Get(ctx, configMapKey, configMap); err != nil { + if apierrors.IsNotFound(err) { + return false, nil + } + return false, errors.Wrapf(err, "failed to check management cluster standby ConfigMap %s", configMapKey.String()) + } + return true, nil +} + +func WrapReconciler(detector Detector, controllerName string, inner reconcile.Reconciler) reconcile.Reconciler { + return &guardedReconciler{ + detector: detector, + controllerName: controllerName, + inner: inner, + } +} + +func WrapWithConfigMapDetector(reader client.Reader, controllerName string, inner reconcile.Reconciler) reconcile.Reconciler { + return WrapReconciler(NewConfigMapDetector(reader), controllerName, inner) +} + +func WrapClusterReconciler(detector Detector, controllerName string, inner reconcile.Reconciler) reconcile.Reconciler { + return &guardedClusterReconciler{ + detector: detector, + controllerName: controllerName, + inner: inner, + } +} + +func WrapClusterReconcilerWithConfigMapDetector(reader client.Reader, controllerName string, inner reconcile.Reconciler) reconcile.Reconciler { + return WrapClusterReconciler(NewConfigMapDetector(reader), controllerName, inner) +} + +func WrapClusterNamedReconciler(detector Detector, reader client.Reader, controllerName string, newObject func() client.Object, getClusterName ClusterNameGetter, inner reconcile.Reconciler) reconcile.Reconciler { + return &guardedClusterNamedReconciler{ + detector: detector, + reader: reader, + controllerName: controllerName, + newObject: newObject, + getClusterName: getClusterName, + inner: inner, + } +} + +func WrapClusterNamedReconcilerWithConfigMapDetector(reader client.Reader, controllerName string, newObject func() client.Object, getClusterName ClusterNameGetter, inner reconcile.Reconciler) reconcile.Reconciler { + return WrapClusterNamedReconciler(NewConfigMapDetector(reader), reader, controllerName, newObject, getClusterName, inner) +} + +func (r *guardedReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + isStandby, err := r.detector.IsStandby(ctx) + if err != nil { + // Fail closed: if standby state cannot be determined, do not mutate resources. + return reconcile.Result{}, err + } + if isStandby { + logSkip(ctx, r.controllerName, "") + return reconcile.Result{RequeueAfter: RequeueAfter}, nil + } + return r.inner.Reconcile(ctx, req) +} + +func (r *guardedClusterReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + isStandby, err := r.detector.IsStandby(ctx) + if err != nil { + return reconcile.Result{}, err + } + if isStandby && req.Name != GlobalClusterName { + logSkip(ctx, r.controllerName, req.Name) + return reconcile.Result{RequeueAfter: RequeueAfter}, nil + } + return r.inner.Reconcile(ctx, req) +} + +func (r *guardedClusterNamedReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { + isStandby, err := r.detector.IsStandby(ctx) + if err != nil { + return reconcile.Result{}, err + } + if !isStandby { + return r.inner.Reconcile(ctx, req) + } + + obj := r.newObject() + if err := r.reader.Get(ctx, req.NamespacedName, obj); err != nil { + if apierrors.IsNotFound(err) { + return r.inner.Reconcile(ctx, req) + } + return reconcile.Result{}, err + } + + clusterName := r.getClusterName(obj) + if clusterName == GlobalClusterName { + return r.inner.Reconcile(ctx, req) + } + logSkip(ctx, r.controllerName, clusterName) + return reconcile.Result{RequeueAfter: RequeueAfter}, nil +} + +func logSkip(ctx context.Context, controllerName, clusterName string) { + logValues := []any{ + "controller", controllerName, + "configMap", klog.KRef(ConfigMapNamespace, ConfigMapName), + "requeueAfter", RequeueAfter, + } + if clusterName != "" { + logValues = append(logValues, "cluster", clusterName) + } + ctrl.LoggerFrom(ctx).V(2).Info("Skipping reconciliation because management cluster is DR standby", logValues...) +} diff --git a/pkg/standby/standby_test.go b/pkg/standby/standby_test.go new file mode 100644 index 000000000000..aca199997306 --- /dev/null +++ b/pkg/standby/standby_test.go @@ -0,0 +1,216 @@ +/* +Copyright 2026 The Kubernetes 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 standby + +import ( + "context" + "errors" + "testing" + + . "github.com/onsi/gomega" + corev1 "k8s.io/api/core/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/types" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +type errorReader struct{} + +func (errorReader) Get(context.Context, client.ObjectKey, client.Object, ...client.GetOption) error { + return errors.New("test error") +} + +func (errorReader) List(context.Context, client.ObjectList, ...client.ListOption) error { + return errors.New("test error") +} + +type fakeDetector struct { + standby bool + err error +} + +func (d fakeDetector) IsStandby(context.Context) (bool, error) { + return d.standby, d.err +} + +type countingReconciler struct { + calls int + res reconcile.Result + err error +} + +func (r *countingReconciler) Reconcile(context.Context, reconcile.Request) (reconcile.Result, error) { + r.calls++ + return r.res, r.err +} + +func TestConfigMapDetectorIsStandby(t *testing.T) { + g := NewWithT(t) + oldNamespace := ConfigMapNamespace + ConfigMapNamespace = "cpaas-system" + defer func() { ConfigMapNamespace = oldNamespace }() + + scheme := runtime.NewScheme() + g.Expect(corev1.AddToScheme(scheme)).To(Succeed()) + + standbyClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{ + Namespace: ConfigMapNamespace, + Name: ConfigMapName, + }, + }).Build() + isStandby, err := NewConfigMapDetector(standbyClient).IsStandby(context.Background()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(isStandby).To(BeTrue()) + + activeClient := fake.NewClientBuilder().WithScheme(scheme).Build() + isStandby, err = NewConfigMapDetector(activeClient).IsStandby(context.Background()) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(isStandby).To(BeFalse()) +} + +func TestConfigMapDetectorReturnsGetError(t *testing.T) { + g := NewWithT(t) + + isStandby, err := NewConfigMapDetector(errorReader{}).IsStandby(context.Background()) + g.Expect(err).To(HaveOccurred()) + g.Expect(isStandby).To(BeFalse()) +} + +func TestWrapReconciler(t *testing.T) { + t.Run("standby skips inner reconciler", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{} + + res, err := WrapReconciler(fakeDetector{standby: true}, "test", inner).Reconcile(context.Background(), reconcile.Request{}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res).To(Equal(reconcile.Result{RequeueAfter: RequeueAfter})) + g.Expect(inner.calls).To(Equal(0)) + }) + + t.Run("active calls inner reconciler", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{res: reconcile.Result{Requeue: true}} + + res, err := WrapReconciler(fakeDetector{}, "test", inner).Reconcile(context.Background(), reconcile.Request{}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res).To(Equal(reconcile.Result{Requeue: true})) + g.Expect(inner.calls).To(Equal(1)) + }) + + t.Run("detector error skips inner reconciler", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{} + detectorErr := apierrors.NewNotFound(schema.GroupResource{Group: "test", Resource: "detectors"}, "test") + + _, err := WrapReconciler(fakeDetector{err: detectorErr}, "test", inner).Reconcile(context.Background(), reconcile.Request{}) + g.Expect(err).To(MatchError(detectorErr)) + g.Expect(inner.calls).To(Equal(0)) + }) +} + +func TestWrapClusterReconciler(t *testing.T) { + t.Run("standby allows global cluster", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{res: reconcile.Result{Requeue: true}} + + res, err := WrapClusterReconciler(fakeDetector{standby: true}, "cluster", inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: GlobalClusterName}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res).To(Equal(reconcile.Result{Requeue: true})) + g.Expect(inner.calls).To(Equal(1)) + }) + + t.Run("standby skips business cluster", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{} + + res, err := WrapClusterReconciler(fakeDetector{standby: true}, "cluster", inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: "business"}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res).To(Equal(reconcile.Result{RequeueAfter: RequeueAfter})) + g.Expect(inner.calls).To(Equal(0)) + }) + + t.Run("active calls inner reconciler", func(t *testing.T) { + g := NewWithT(t) + inner := &countingReconciler{} + + _, err := WrapClusterReconciler(fakeDetector{}, "cluster", inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Name: "business"}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(inner.calls).To(Equal(1)) + }) +} + +func TestWrapClusterNamedReconciler(t *testing.T) { + scheme := runtime.NewScheme() + NewWithT(t).Expect(corev1.AddToScheme(scheme)).To(Succeed()) + + newObject := func() client.Object { return &corev1.ConfigMap{} } + getClusterName := func(obj client.Object) string { return obj.(*corev1.ConfigMap).Data["clusterName"] } + + t.Run("standby allows object from global cluster", func(t *testing.T) { + g := NewWithT(t) + reader := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Namespace: metav1.NamespaceDefault, Name: "cm"}, + Data: map[string]string{"clusterName": GlobalClusterName}, + }).Build() + inner := &countingReconciler{} + + _, err := WrapClusterNamedReconciler(fakeDetector{standby: true}, reader, "test", newObject, getClusterName, inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: "cm"}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(inner.calls).To(Equal(1)) + }) + + t.Run("standby skips object from business cluster", func(t *testing.T) { + g := NewWithT(t) + reader := fake.NewClientBuilder().WithScheme(scheme).WithObjects(&corev1.ConfigMap{ + ObjectMeta: metav1.ObjectMeta{Namespace: metav1.NamespaceDefault, Name: "cm"}, + Data: map[string]string{"clusterName": "business"}, + }).Build() + inner := &countingReconciler{} + + res, err := WrapClusterNamedReconciler(fakeDetector{standby: true}, reader, "test", newObject, getClusterName, inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: "cm"}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(res).To(Equal(reconcile.Result{RequeueAfter: RequeueAfter})) + g.Expect(inner.calls).To(Equal(0)) + }) + + t.Run("standby lets not found object reach inner reconciler", func(t *testing.T) { + g := NewWithT(t) + reader := fake.NewClientBuilder().WithScheme(scheme).Build() + inner := &countingReconciler{} + + _, err := WrapClusterNamedReconciler(fakeDetector{standby: true}, reader, "test", newObject, getClusterName, inner).Reconcile(context.Background(), reconcile.Request{NamespacedName: types.NamespacedName{Namespace: metav1.NamespaceDefault, Name: "missing"}}) + g.Expect(err).NotTo(HaveOccurred()) + g.Expect(inner.calls).To(Equal(1)) + }) + + t.Run("detector error skips inner reconciler", func(t *testing.T) { + g := NewWithT(t) + reader := fake.NewClientBuilder().WithScheme(scheme).Build() + inner := &countingReconciler{} + detectorErr := errors.New("detector error") + + _, err := WrapClusterNamedReconciler(fakeDetector{err: detectorErr}, reader, "test", newObject, getClusterName, inner).Reconcile(context.Background(), reconcile.Request{}) + g.Expect(err).To(MatchError(detectorErr)) + g.Expect(inner.calls).To(Equal(0)) + }) +}