KCP: Extend rollout logic for in-place updates

Signed-off-by: Stefan Büringer [email protected]

Author: sbueringer

controlplane/kubeadm/internal/controllers/controller.go

@@ -95,6 +95,11 @@ type KubeadmControlPlaneReconciler struct {
 	managementCluster         internal.ManagementCluster
 	managementClusterUncached internal.ManagementCluster
 	ssaCache                  ssa.Cache
+
+	// Only used for testing
+	overrideTryInPlaceUpdateFunc      func(ctx context.Context, controlPlane *internal.ControlPlane, machineToInPlaceUpdate *clusterv1.Machine, machinesNeedingRolloutResult internal.NotUpToDateResult) (bool, ctrl.Result, error)
+	overrideScaleUpControlPlaneFunc   func(ctx context.Context, controlPlane *internal.ControlPlane) (ctrl.Result, error)
+	overrideScaleDownControlPlaneFunc func(ctx context.Context, controlPlane *internal.ControlPlane, machineToDelete *clusterv1.Machine) (ctrl.Result, error)
 }
 
 func (r *KubeadmControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
@@ -476,7 +481,7 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, controlPl
 		}
 		log.Info(fmt.Sprintf("Rolling out Control Plane machines: %s", strings.Join(allMessages, ",")), "machinesNeedingRollout", machinesNeedingRollout.Names())
 		v1beta1conditions.MarkFalse(controlPlane.KCP, controlplanev1.MachinesSpecUpToDateV1Beta1Condition, controlplanev1.RollingUpdateInProgressV1Beta1Reason, clusterv1.ConditionSeverityWarning, "Rolling %d replicas with outdated spec (%d replicas up to date)", len(machinesNeedingRollout), len(controlPlane.Machines)-len(machinesNeedingRollout))
-		return r.upgradeControlPlane(ctx, controlPlane, machinesNeedingRollout)
+		return r.updateControlPlane(ctx, controlPlane, machinesNeedingRollout, machinesNeedingRolloutResults)
 	default:
 		// make sure last upgrade operation is marked as completed.
 		// NOTE: we are checking the condition already exists in order to avoid to set this condition at the first
@@ -506,7 +511,12 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, controlPl
 	case numMachines > desiredReplicas:
 		log.Info("Scaling down control plane", "desired", desiredReplicas, "existing", numMachines)
 		// The last parameter (i.e. machines needing to be rolled out) should always be empty here.
-		return r.scaleDownControlPlane(ctx, controlPlane, collections.Machines{})
+		// Pick the Machine that we should scale down.
+		machineToDelete, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, controlPlane, collections.Machines{})
+		if err != nil {
+			return ctrl.Result{}, errors.Wrap(err, "failed to select machine for scale down")
+		}
+		return r.scaleDownControlPlane(ctx, controlPlane, machineToDelete)
 	}
 
 	// Get the workload cluster client.

controlplane/kubeadm/internal/controllers/inplace.go

@@ -0,0 +1,40 @@
+/*
+Copyright 2025 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 controllers
+
+import (
+	"context"
+
+	"github.com/pkg/errors"
+	ctrl "sigs.k8s.io/controller-runtime"
+
+	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
+	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal"
+)
+
+func (r *KubeadmControlPlaneReconciler) tryInPlaceUpdate(
+	ctx context.Context,
+	controlPlane *internal.ControlPlane,
+	machineToInPlaceUpdate *clusterv1.Machine,
+	machinesNeedingRolloutResult internal.NotUpToDateResult,
+) (fallbackToScaleDown bool, _ ctrl.Result, _ error) {
+	if r.overrideTryInPlaceUpdateFunc != nil {
+		return r.overrideTryInPlaceUpdateFunc(ctx, controlPlane, machineToInPlaceUpdate, machinesNeedingRolloutResult)
+	}
+
+	return false, ctrl.Result{}, errors.Errorf("in-place updates is not implemented yet")
+}

controlplane/kubeadm/internal/controllers/scale.go

@@ -63,6 +63,10 @@ func (r *KubeadmControlPlaneReconciler) initializeControlPlane(ctx context.Conte
 }
 
 func (r *KubeadmControlPlaneReconciler) scaleUpControlPlane(ctx context.Context, controlPlane *internal.ControlPlane) (ctrl.Result, error) {
+	if r.overrideScaleUpControlPlaneFunc != nil {
+		return r.overrideScaleUpControlPlaneFunc(ctx, controlPlane)
+	}
+
 	log := ctrl.LoggerFrom(ctx)
 
 	// Run preflight checks to ensure that the control plane is stable before proceeding with a scale up/scale down operation; if not, wait.
@@ -95,16 +99,14 @@ func (r *KubeadmControlPlaneReconciler) scaleUpControlPlane(ctx context.Context,
 func (r *KubeadmControlPlaneReconciler) scaleDownControlPlane(
 	ctx context.Context,
 	controlPlane *internal.ControlPlane,
-	outdatedMachines collections.Machines,
+	machineToDelete *clusterv1.Machine,
 ) (ctrl.Result, error) {
-	log := ctrl.LoggerFrom(ctx)
-
-	// Pick the Machine that we should scale down.
-	machineToDelete, err := selectMachineForScaleDown(ctx, controlPlane, outdatedMachines)
-	if err != nil {
-		return ctrl.Result{}, errors.Wrap(err, "failed to select machine for scale down")
+	if r.overrideScaleDownControlPlaneFunc != nil {
+		return r.overrideScaleDownControlPlaneFunc(ctx, controlPlane, machineToDelete)
 	}
 
+	log := ctrl.LoggerFrom(ctx)
+
 	// Run preflight checks ensuring the control plane is stable before proceeding with a scale up/scale down operation; if not, wait.
 	// Given that we're scaling down, we can exclude the machineToDelete from the preflight checks.
 	if result, err := r.preflightChecks(ctx, controlPlane, machineToDelete); err != nil || !result.IsZero() {
@@ -265,7 +267,8 @@ func preflightCheckCondition(kind string, obj *clusterv1.Machine, conditionType
 	return nil
 }
 
-// selectMachineForScaleDown select a machine candidate for scaling down. The selection is a two phase process:
+// selectMachineForInPlaceUpdateOrScaleDown select a machine candidate for scaling down or for in-place update.
+// The selection is a two phase process:
 //
 // In the first phase it selects a subset of machines eligible for deletion:
 // - if there are outdated machines with the delete machine annotation, use them as eligible subset (priority to user requests, part 1)
@@ -276,18 +279,20 @@ func preflightCheckCondition(kind string, obj *clusterv1.Machine, conditionType
 //
 // Once the subset of machines eligible for deletion is identified, one machine is picked out of this subset by
 // selecting the machine in the failure domain with most machines (including both eligible and not eligible machines).
-func selectMachineForScaleDown(ctx context.Context, controlPlane *internal.ControlPlane, outdatedMachines collections.Machines) (*clusterv1.Machine, error) {
+func selectMachineForInPlaceUpdateOrScaleDown(ctx context.Context, controlPlane *internal.ControlPlane, outdatedMachines collections.Machines) (*clusterv1.Machine, error) {
 	// Select the subset of machines eligible for scale down.
-	eligibleMachines := controlPlane.Machines
+	var eligibleMachines collections.Machines
 	switch {
 	case controlPlane.MachineWithDeleteAnnotation(outdatedMachines).Len() > 0:
 		eligibleMachines = controlPlane.MachineWithDeleteAnnotation(outdatedMachines)
-	case controlPlane.MachineWithDeleteAnnotation(eligibleMachines).Len() > 0:
-		eligibleMachines = controlPlane.MachineWithDeleteAnnotation(eligibleMachines)
+	case controlPlane.MachineWithDeleteAnnotation(controlPlane.Machines).Len() > 0:
+		eligibleMachines = controlPlane.MachineWithDeleteAnnotation(controlPlane.Machines)
 	case controlPlane.UnhealthyMachinesWithUnhealthyControlPlaneComponents(outdatedMachines).Len() > 0:
 		eligibleMachines = controlPlane.UnhealthyMachinesWithUnhealthyControlPlaneComponents(outdatedMachines)
 	case outdatedMachines.Len() > 0:
 		eligibleMachines = outdatedMachines
+	default:
+		eligibleMachines = controlPlane.Machines
 	}
 
 	// Pick an eligible machine from the failure domain with most machines in (including both eligible and not eligible machines)

controlplane/kubeadm/internal/controllers/scale_test.go

@@ -284,7 +284,9 @@ func TestKubeadmControlPlaneReconciler_scaleDownControlPlane_NoError(t *testing.
 		}
 		controlPlane.InjectTestManagementCluster(r.managementCluster)
 
-		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, controlPlane.Machines)
+		machineToDelete, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, controlPlane, controlPlane.Machines)
+		g.Expect(err).ToNot(HaveOccurred())
+		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, machineToDelete)
 		g.Expect(err).ToNot(HaveOccurred())
 		g.Expect(result).To(BeComparableTo(ctrl.Result{Requeue: true}))
 
@@ -326,7 +328,9 @@ func TestKubeadmControlPlaneReconciler_scaleDownControlPlane_NoError(t *testing.
 		}
 		controlPlane.InjectTestManagementCluster(r.managementCluster)
 
-		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, controlPlane.Machines)
+		machineToDelete, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, controlPlane, controlPlane.Machines)
+		g.Expect(err).ToNot(HaveOccurred())
+		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, machineToDelete)
 		g.Expect(err).ToNot(HaveOccurred())
 		g.Expect(result).To(BeComparableTo(ctrl.Result{Requeue: true}))
 
@@ -364,7 +368,9 @@ func TestKubeadmControlPlaneReconciler_scaleDownControlPlane_NoError(t *testing.
 		}
 		controlPlane.InjectTestManagementCluster(r.managementCluster)
 
-		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, controlPlane.Machines)
+		machineToDelete, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, controlPlane, controlPlane.Machines)
+		g.Expect(err).ToNot(HaveOccurred())
+		result, err := r.scaleDownControlPlane(context.Background(), controlPlane, machineToDelete)
 		g.Expect(err).ToNot(HaveOccurred())
 		g.Expect(result).To(BeComparableTo(ctrl.Result{RequeueAfter: preflightFailedRequeueAfter}))
 
@@ -374,7 +380,7 @@ func TestKubeadmControlPlaneReconciler_scaleDownControlPlane_NoError(t *testing.
 	})
 }
 
-func TestSelectMachineForScaleDown(t *testing.T) {
+func TestSelectMachineForInPlaceUpdateOrScaleDown(t *testing.T) {
 	kcp := controlplanev1.KubeadmControlPlane{
 		Spec: controlplanev1.KubeadmControlPlaneSpec{},
 	}
@@ -503,7 +509,7 @@ func TestSelectMachineForScaleDown(t *testing.T) {
 		t.Run(tc.name, func(t *testing.T) {
 			g := NewWithT(t)
 
-			selectedMachine, err := selectMachineForScaleDown(ctx, tc.cp, tc.outDatedMachines)
+			selectedMachine, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, tc.cp, tc.outDatedMachines)
 
 			if tc.expectErr {
 				g.Expect(err).To(HaveOccurred())

controlplane/kubeadm/internal/controllers/upgrade.go

@@ -27,13 +27,15 @@ import (
 	bootstrapv1 "sigs.k8s.io/cluster-api/api/bootstrap/kubeadm/v1beta2"
 	controlplanev1 "sigs.k8s.io/cluster-api/api/controlplane/kubeadm/v1beta2"
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal"
+	"sigs.k8s.io/cluster-api/feature"
 	"sigs.k8s.io/cluster-api/util/collections"
 )
 
-func (r *KubeadmControlPlaneReconciler) upgradeControlPlane(
+func (r *KubeadmControlPlaneReconciler) updateControlPlane(
 	ctx context.Context,
 	controlPlane *internal.ControlPlane,
-	machinesRequireUpgrade collections.Machines,
+	machinesNeedingRollout collections.Machines,
+	machinesNeedingRolloutResults map[string]internal.NotUpToDateResult,
 ) (ctrl.Result, error) {
 	log := ctrl.LoggerFrom(ctx)
 
@@ -42,17 +44,17 @@ func (r *KubeadmControlPlaneReconciler) upgradeControlPlane(
 	workloadCluster, err := controlPlane.GetWorkloadCluster(ctx)
 	if err != nil {
 		log.Error(err, "failed to get remote client for workload cluster", "Cluster", klog.KObj(controlPlane.Cluster))
-		return ctrl.Result{}, err
+		return ctrl.Result{}, errors.Wrapf(err, "failed to update control plane")
 	}
 
 	parsedVersion, err := semver.ParseTolerant(controlPlane.KCP.Spec.Version)
 	if err != nil {
-		return ctrl.Result{}, errors.Wrapf(err, "failed to parse kubernetes version %q", controlPlane.KCP.Spec.Version)
+		return ctrl.Result{}, errors.Wrapf(err, "failed to update control plane: failed to parse Kubernetes version %q", controlPlane.KCP.Spec.Version)
 	}
 
 	// Ensure kubeadm clusterRoleBinding for v1.29+ as per https://github.com/kubernetes/kubernetes/pull/121305
 	if err := workloadCluster.AllowClusterAdminPermissions(ctx, parsedVersion); err != nil {
-		return ctrl.Result{}, errors.Wrap(err, "failed to set cluster-admin ClusterRoleBinding for kubeadm")
+		return ctrl.Result{}, errors.Wrap(err, "failed to update control plane: failed to set cluster-admin ClusterRoleBinding for kubeadm")
 	}
 
 	kubeadmCMMutators := make([]func(*bootstrapv1.ClusterConfiguration), 0)
@@ -81,21 +83,75 @@ func (r *KubeadmControlPlaneReconciler) upgradeControlPlane(
 
 	// collectively update Kubeadm config map
 	if err = workloadCluster.UpdateClusterConfiguration(ctx, parsedVersion, kubeadmCMMutators...); err != nil {
-		return ctrl.Result{}, err
+		return ctrl.Result{}, errors.Wrapf(err, "failed to update control plane")
 	}
 
 	switch controlPlane.KCP.Spec.Rollout.Strategy.Type {
 	case controlplanev1.RollingUpdateStrategyType:
 		// RolloutStrategy is currently defaulted and validated to be RollingUpdate
-		// We can ignore MaxUnavailable because we are enforcing health checks before we get here.
-		maxNodes := *controlPlane.KCP.Spec.Replicas + int32(controlPlane.KCP.Spec.Rollout.Strategy.RollingUpdate.MaxSurge.IntValue())
-		if int32(controlPlane.Machines.Len()) < maxNodes {
-			// scaleUp ensures that we don't continue scaling up while waiting for Machines to have NodeRefs
-			return r.scaleUpControlPlane(ctx, controlPlane)
+		res, err := r.rollingUpdate(ctx, controlPlane, machinesNeedingRollout, machinesNeedingRolloutResults)
+		if err != nil {
+			return ctrl.Result{}, errors.Wrapf(err, "failed to update control plane")
 		}
-		return r.scaleDownControlPlane(ctx, controlPlane, machinesRequireUpgrade)
+		return res, nil
 	default:
 		log.Info("RolloutStrategy type is not set to RollingUpdate, unable to determine the strategy for rolling out machines")
 		return ctrl.Result{}, nil
 	}
 }
+
+func (r *KubeadmControlPlaneReconciler) rollingUpdate(
+	ctx context.Context,
+	controlPlane *internal.ControlPlane,
+	machinesNeedingRollout collections.Machines,
+	machinesNeedingRolloutResults map[string]internal.NotUpToDateResult,
+) (ctrl.Result, error) {
+	currentReplicas := int32(controlPlane.Machines.Len())
+	currentUpToDateReplicas := int32(len(controlPlane.UpToDateMachines()))
+	desiredReplicas := *controlPlane.KCP.Spec.Replicas
+	maxSurge := int32(controlPlane.KCP.Spec.Rollout.Strategy.RollingUpdate.MaxSurge.IntValue())
+	// Note: As MaxSurge is validated to be either 0 or 1, maxReplicas will be either desiredReplicas or desiredReplicas+1
+	maxReplicas := desiredReplicas + maxSurge
+
+	// If currentReplicas < maxReplicas we have to scale up
+	// Note: This is done to ensure we have as many Machines as allowed during rollout to maximize fault tolerance.
+	if currentReplicas < maxReplicas {
+		// Note: scaleUpControlPlane ensures that we don't continue scale up while waiting for Machines to have NodeRefs.
+		return r.scaleUpControlPlane(ctx, controlPlane)
+	}
+
+	// If currentReplicas >= maxReplicas we have to scale down
+	// Note: If we are already at or above the maximum Machines we have to in-place update or delete a Machine
+	// to make progress with the update (as we cannot create additional new Machines above the maximum).
+
+	// Pick the Machine that we should in-place update or scale down.
+	machineToInPlaceUpdateOrScaleDown, err := selectMachineForInPlaceUpdateOrScaleDown(ctx, controlPlane, machinesNeedingRollout)
+	if err != nil {
+		return ctrl.Result{}, errors.Wrap(err, "failed to select next Machine for rollout")
+	}
+	machinesNeedingRolloutResult, ok := machinesNeedingRolloutResults[machineToInPlaceUpdateOrScaleDown.Name]
+	if !ok {
+		// Note: This should never happen as we store results for all Machines in machinesNeedingRolloutResults.
+		return ctrl.Result{}, errors.Errorf("failed to check if Machine is UpToDate %s", machineToInPlaceUpdateOrScaleDown.Name)
+	}
+
+	// If the selected Machine is eligible for in-place update and we don't already have enough up-to-date replicas, try in-place update.
+	// Note: To be safe we only try an in-place update when we would otherwise delete a Machine. This ensures we could
+	// afford if the in-place update fails and the Machine becomes unavailable (and eventually MHC kicks in and the Machine is recreated).
+	if feature.Gates.Enabled(feature.InPlaceUpdates) &&
+		machinesNeedingRolloutResult.EligibleForInPlaceUpdate && currentUpToDateReplicas < desiredReplicas {
+		fallbackToScaleDown, res, err := r.tryInPlaceUpdate(ctx, controlPlane, machineToInPlaceUpdateOrScaleDown, machinesNeedingRolloutResult)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+		if !res.IsZero() {
+			return res, nil
+		}
+		if fallbackToScaleDown {
+			return r.scaleDownControlPlane(ctx, controlPlane, machineToInPlaceUpdateOrScaleDown)
+		}
+		// In-place update triggered
+		return ctrl.Result{}, nil // Note: Requeue is not needed, changes to Machines trigger another reconcile.
+	}
+	return r.scaleDownControlPlane(ctx, controlPlane, machineToInPlaceUpdateOrScaleDown)
+}