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