✨ Introduce reconciler rate-limiting and hook caching

.github/workflows/pr-golangci-lint.yaml

@@ -11,6 +11,10 @@ jobs:
   golangci:
     name: lint
     runs-on: ubuntu-latest
+    env:
+      # Required with Go 1.24 to enable usage of synctest in unit tests
+      # Can be removed after we bumped to Go 1.25.
+      GOEXPERIMENT: synctest
     strategy:
       fail-fast: false
       matrix:

.golangci.yml

@@ -20,6 +20,7 @@ linters:
     - durationcheck # multiplying two durations
     - errcheck # unchecked errors
     - errchkjson # invalid types passed to json encoder
+    - forbidigo # allows to block usage of funcs
     - ginkgolinter # ginkgo and gomega
     - gocritic # bugs, performance, style (we could add custom ones to this one)
     - godot # checks that comments end in a period
@@ -52,6 +53,10 @@ linters:
     # TODO: It will be dropped when the Go version migration is done.
     - usetesting
   settings:
+    forbidigo:
+      forbid:
+        - pattern: ctrl.NewControllerManagedBy
+          msg: Use capicontrollerutil.NewControllerManagedBy instead
     ginkgolinter:
       forbid-focus-container: true
     gocritic:
@@ -180,6 +185,8 @@ linters:
           alias: topologynames
         - pkg: sigs.k8s.io/cluster-api/internal/util/client
           alias: "clientutil"
+        - pkg: sigs.k8s.io/cluster-api/internal/util/controller
+          alias: "capicontrollerutil"
         # CAPD
         - pkg: sigs.k8s.io/cluster-api/test/infrastructure/docker/api/v1alpha3
           alias: infrav1alpha3

Makefile

@@ -31,6 +31,10 @@ GO_CONTAINER_IMAGE ?= docker.io/library/golang:$(GO_VERSION)
 GOTOOLCHAIN = go$(GO_VERSION)
 export GOTOOLCHAIN
 
+# Required with Go 1.24 to enable usage of synctest in unit tests
+# Can be removed after we bumped to Go 1.25.
+export GOEXPERIMENT=synctest
+
 # Use GOPROXY environment variable if set
 GOPROXY := $(shell go env GOPROXY)
 ifeq ($(GOPROXY),)

bootstrap/kubeadm/config/manager/manager.yaml

@@ -22,7 +22,7 @@ spec:
             - "--leader-elect"
             - "--diagnostics-address=${CAPI_DIAGNOSTICS_ADDRESS:=:8443}"
             - "--insecure-diagnostics=${CAPI_INSECURE_DIAGNOSTICS:=false}"
-            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},KubeadmBootstrapFormatIgnition=${EXP_KUBEADM_BOOTSTRAP_FORMAT_IGNITION:=false},PriorityQueue=${EXP_PRIORITY_QUEUE:=false}"
+            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},KubeadmBootstrapFormatIgnition=${EXP_KUBEADM_BOOTSTRAP_FORMAT_IGNITION:=false},PriorityQueue=${EXP_PRIORITY_QUEUE:=false},ReconcilerRateLimiting=${EXP_RECONCILER_RATE_LIMITING:=false}"
             - "--bootstrap-token-ttl=${KUBEADM_BOOTSTRAP_TOKEN_TTL:=15m}"
           image: controller:latest
           name: manager

bootstrap/kubeadm/internal/controllers/kubeadmconfig_controller.go

@@ -37,7 +37,6 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/handler"
@@ -53,6 +52,7 @@ import (
 	bsutil "sigs.k8s.io/cluster-api/bootstrap/util"
 	"sigs.k8s.io/cluster-api/controllers/clustercache"
 	"sigs.k8s.io/cluster-api/feature"
+	capicontrollerutil "sigs.k8s.io/cluster-api/internal/util/controller"
 	"sigs.k8s.io/cluster-api/internal/util/taints"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/conditions"
@@ -116,33 +116,26 @@ func (r *KubeadmConfigReconciler) SetupWithManager(ctx context.Context, mgr ctrl
 	}
 
 	predicateLog := ctrl.LoggerFrom(ctx).WithValues("controller", "kubeadmconfig")
-	b := ctrl.NewControllerManagedBy(mgr).
+	b := capicontrollerutil.NewControllerManagedBy(mgr, predicateLog).
 		For(&bootstrapv1.KubeadmConfig{}).
 		WithOptions(options).
 		Watches(
 			&clusterv1.Machine{},
 			handler.EnqueueRequestsFromMapFunc(r.MachineToBootstrapMapFunc),
-			builder.WithPredicates(predicates.ResourceIsChanged(mgr.GetScheme(), predicateLog)),
 		).WithEventFilter(predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue))
 
 	if feature.Gates.Enabled(feature.MachinePool) {
 		b = b.Watches(
 			&clusterv1.MachinePool{},
 			handler.EnqueueRequestsFromMapFunc(r.MachinePoolToBootstrapMapFunc),
-			builder.WithPredicates(predicates.ResourceIsChanged(mgr.GetScheme(), predicateLog)),
 		)
 	}
 
 	b = b.Watches(
 		&clusterv1.Cluster{},
 		handler.EnqueueRequestsFromMapFunc(r.ClusterToKubeadmConfigs),
-		builder.WithPredicates(
-			predicates.All(mgr.GetScheme(), predicateLog,
-				predicates.ResourceIsChanged(mgr.GetScheme(), predicateLog),
-				predicates.ClusterPausedTransitionsOrInfrastructureProvisioned(mgr.GetScheme(), predicateLog),
-				predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue),
-			),
-		),
+		predicates.ClusterPausedTransitionsOrInfrastructureProvisioned(mgr.GetScheme(), predicateLog),
+		predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue),
 	).WatchesRawSource(r.ClusterCache.GetClusterSource("kubeadmconfig", r.ClusterToKubeadmConfigs))
 
 	if err := b.Complete(r); err != nil {
@@ -268,12 +261,6 @@ func (r *KubeadmConfigReconciler) Reconcile(ctx context.Context, req ctrl.Reques
 		if err := patchHelper.Patch(ctx, config, patchOpts...); err != nil {
 			rerr = kerrors.NewAggregate([]error{rerr, err})
 		}
-
-		// Note: controller-runtime logs a warning that non-empty result is ignored
-		// if error is not nil, so setting result here to empty to avoid noisy warnings.
-		if rerr != nil {
-			retRes = ctrl.Result{}
-		}
 	}()
 
 	// Ignore deleted KubeadmConfigs.

config/manager/manager.yaml

@@ -23,7 +23,7 @@ spec:
             - "--leader-elect"
             - "--diagnostics-address=${CAPI_DIAGNOSTICS_ADDRESS:=:8443}"
             - "--insecure-diagnostics=${CAPI_INSECURE_DIAGNOSTICS:=false}"
-            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},ClusterTopology=${CLUSTER_TOPOLOGY:=false},RuntimeSDK=${EXP_RUNTIME_SDK:=false},MachineSetPreflightChecks=${EXP_MACHINE_SET_PREFLIGHT_CHECKS:=true},MachineWaitForVolumeDetachConsiderVolumeAttachments=${EXP_MACHINE_WAITFORVOLUMEDETACH_CONSIDER_VOLUMEATTACHMENTS:=true},PriorityQueue=${EXP_PRIORITY_QUEUE:=false},InPlaceUpdates=${EXP_IN_PLACE_UPDATES:=false},MachineTaintPropagation=${EXP_MACHINE_TAINT_PROPAGATION:=false}"
+            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},ClusterTopology=${CLUSTER_TOPOLOGY:=false},RuntimeSDK=${EXP_RUNTIME_SDK:=false},MachineSetPreflightChecks=${EXP_MACHINE_SET_PREFLIGHT_CHECKS:=true},MachineWaitForVolumeDetachConsiderVolumeAttachments=${EXP_MACHINE_WAITFORVOLUMEDETACH_CONSIDER_VOLUMEATTACHMENTS:=true},PriorityQueue=${EXP_PRIORITY_QUEUE:=false},ReconcilerRateLimiting=${EXP_RECONCILER_RATE_LIMITING:=false},InPlaceUpdates=${EXP_IN_PLACE_UPDATES:=false},MachineTaintPropagation=${EXP_MACHINE_TAINT_PROPAGATION:=false}"
           image: controller:latest
           name: manager
           env:

controllers/clustercache/cluster_cache.go

@@ -44,6 +44,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/source"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
+	capicontrollerutil "sigs.k8s.io/cluster-api/internal/util/controller"
 	"sigs.k8s.io/cluster-api/util/predicates"
 )
 
@@ -328,7 +329,8 @@ func SetupWithManager(ctx context.Context, mgr manager.Manager, options Options,
 		cacheCtxCancel:        cacheCtxCancel,
 	}
 
-	err := ctrl.NewControllerManagedBy(mgr).
+	predicateLog := ctrl.LoggerFrom(ctx).WithValues("controller", "clustercache")
+	err := capicontrollerutil.NewControllerManagedBy(mgr, predicateLog).
 		Named("clustercache").
 		For(&clusterv1.Cluster{}).
 		WithOptions(controllerOptions).

controllers/crdmigrator/crd_migrator.go

@@ -45,6 +45,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
+	capicontrollerutil "sigs.k8s.io/cluster-api/internal/util/controller"
 	"sigs.k8s.io/cluster-api/util/cache"
 	"sigs.k8s.io/cluster-api/util/contract"
 	"sigs.k8s.io/cluster-api/util/predicates"
@@ -121,7 +122,7 @@ func (r *CRDMigrator) SetupWithManager(ctx context.Context, mgr ctrl.Manager, co
 	}
 
 	predicateLog := ctrl.LoggerFrom(ctx).WithValues("controller", "crdmigrator")
-	err := ctrl.NewControllerManagedBy(mgr).
+	err := capicontrollerutil.NewControllerManagedBy(mgr, predicateLog).
 		For(&apiextensionsv1.CustomResourceDefinition{},
 			// This controller uses a PartialObjectMetadata watch/informer to avoid an informer for CRDs
 			// to reduce memory usage.

controlplane/kubeadm/config/manager/manager.yaml

@@ -22,7 +22,7 @@ spec:
             - "--leader-elect"
             - "--diagnostics-address=${CAPI_DIAGNOSTICS_ADDRESS:=:8443}"
             - "--insecure-diagnostics=${CAPI_INSECURE_DIAGNOSTICS:=false}"
-            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},ClusterTopology=${CLUSTER_TOPOLOGY:=false},KubeadmBootstrapFormatIgnition=${EXP_KUBEADM_BOOTSTRAP_FORMAT_IGNITION:=false},PriorityQueue=${EXP_PRIORITY_QUEUE:=false},InPlaceUpdates=${EXP_IN_PLACE_UPDATES:=false}"
+            - "--feature-gates=MachinePool=${EXP_MACHINE_POOL:=true},ClusterTopology=${CLUSTER_TOPOLOGY:=false},KubeadmBootstrapFormatIgnition=${EXP_KUBEADM_BOOTSTRAP_FORMAT_IGNITION:=false},PriorityQueue=${EXP_PRIORITY_QUEUE:=false},ReconcilerRateLimiting=${EXP_RECONCILER_RATE_LIMITING:=false},InPlaceUpdates=${EXP_IN_PLACE_UPDATES:=false}"
           image: controller:latest
           name: manager
           env:

controlplane/kubeadm/internal/controllers/controller.go

@@ -36,7 +36,6 @@ import (
 	"k8s.io/klog/v2"
 	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/controller"
 	"sigs.k8s.io/controller-runtime/pkg/controller/controllerutil"
@@ -49,6 +48,7 @@ import (
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal"
 	runtimeclient "sigs.k8s.io/cluster-api/exp/runtime/client"
 	"sigs.k8s.io/cluster-api/feature"
+	capicontrollerutil "sigs.k8s.io/cluster-api/internal/util/controller"
 	"sigs.k8s.io/cluster-api/internal/util/inplace"
 	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util"
@@ -85,7 +85,7 @@ type KubeadmControlPlaneReconciler struct {
 	APIReader           client.Reader
 	SecretCachingClient client.Client
 	RuntimeClient       runtimeclient.Client
-	controller          controller.Controller
+	controller          capicontrollerutil.Controller
 	recorder            record.EventRecorder
 	ClusterCache        clustercache.ClusterCache
 
@@ -132,23 +132,18 @@ func (r *KubeadmControlPlaneReconciler) SetupWithManager(ctx context.Context, mg
 	}
 
 	predicateLog := ctrl.LoggerFrom(ctx).WithValues("controller", "kubeadmcontrolplane")
-	c, err := ctrl.NewControllerManagedBy(mgr).
+	c, err := capicontrollerutil.NewControllerManagedBy(mgr, predicateLog).
 		For(&controlplanev1.KubeadmControlPlane{}).
-		Owns(&clusterv1.Machine{}, builder.WithPredicates(predicates.ResourceIsChanged(mgr.GetScheme(), predicateLog))).
+		Owns(&clusterv1.Machine{}).
 		WithOptions(options).
 		WithEventFilter(predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue)).
 		Watches(
 			&clusterv1.Cluster{},
 			handler.EnqueueRequestsFromMapFunc(r.ClusterToKubeadmControlPlane),
-			builder.WithPredicates(
-				predicates.All(mgr.GetScheme(), predicateLog,
-					predicates.ResourceIsChanged(mgr.GetScheme(), predicateLog),
-					predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue),
-					predicates.Any(mgr.GetScheme(), predicateLog,
-						predicates.ClusterPausedTransitionsOrInfrastructureProvisioned(mgr.GetScheme(), predicateLog),
-						predicates.ClusterTopologyVersionChanged(mgr.GetScheme(), predicateLog),
-					),
-				),
+			predicates.ResourceHasFilterLabel(mgr.GetScheme(), predicateLog, r.WatchFilterValue),
+			predicates.Any(mgr.GetScheme(), predicateLog,
+				predicates.ClusterPausedTransitionsOrInfrastructureProvisioned(mgr.GetScheme(), predicateLog),
+				predicates.ClusterTopologyVersionChanged(mgr.GetScheme(), predicateLog),
 			),
 		).
 		WatchesRawSource(r.ClusterCache.GetClusterSource("kubeadmcontrolplane", r.ClusterToKubeadmControlPlane,
@@ -283,12 +278,6 @@ func (r *KubeadmControlPlaneReconciler) Reconcile(ctx context.Context, req ctrl.
 				res = ctrl.Result{RequeueAfter: 20 * time.Second}
 			}
 		}
-
-		// Note: controller-runtime logs a warning that non-empty result is ignored
-		// if error is not nil, so setting result here to empty to avoid noisy warnings.
-		if reterr != nil {
-			res = ctrl.Result{}
-		}
 	}()
 
 	if !kcp.DeletionTimestamp.IsZero() {

controlplane/kubeadm/internal/controllers/scale.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"fmt"
 	"strings"
+	"time"
 
 	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
@@ -190,6 +191,9 @@ func (r *KubeadmControlPlaneReconciler) preflightChecks(ctx context.Context, con
 			}
 			log.Info(fmt.Sprintf("Waiting for a version upgrade to %s to be propagated", v))
 			controlPlane.PreflightCheckResults.TopologyVersionMismatch = true
+			// Slow down reconcile frequency, as deferring a version upgrade waits for slow processes,
+			// e.g. workers are completing a previous upgrade step.
+			r.controller.DeferNextReconcileForObject(controlPlane.KCP, time.Now().Add(5*time.Second))
 			return ctrl.Result{RequeueAfter: preflightFailedRequeueAfter}
 		}
 	}
@@ -198,6 +202,8 @@ func (r *KubeadmControlPlaneReconciler) preflightChecks(ctx context.Context, con
 	if controlPlane.HasDeletingMachine() {
 		controlPlane.PreflightCheckResults.HasDeletingMachine = true
 		log.Info("Waiting for machines to be deleted", "machines", strings.Join(controlPlane.Machines.Filter(collections.HasDeletionTimestamp).Names(), ", "))
+		// Slow down reconcile frequency, deletion is a slow process.
+		r.controller.DeferNextReconcileForObject(controlPlane.KCP, time.Now().Add(5*time.Second))
 		return ctrl.Result{RequeueAfter: deleteRequeueAfter}
 	}
 
@@ -254,7 +260,10 @@ loopmachines:
 		r.recorder.Eventf(controlPlane.KCP, corev1.EventTypeWarning, "ControlPlaneUnhealthy",
 			"Waiting for control plane to pass preflight checks to continue reconciliation: %v", aggregatedError)
 		log.Info("Waiting for control plane to pass preflight checks", "failures", aggregatedError.Error())
-
+		// Slow down reconcile frequency, it takes some time before control plane components stabilize
+		// after a new Machine is created. Similarly, if there are issues on running Machines, it
+		// usually takes some time to get back to normal state.
+		r.controller.DeferNextReconcileForObject(controlPlane.KCP, time.Now().Add(5*time.Second))
 		return ctrl.Result{RequeueAfter: preflightFailedRequeueAfter}
 	}