✨ KCP: implement trigger in-place update

controlplane/kubeadm/internal/control_plane.go

@@ -33,8 +33,10 @@ import (
 	bootstrapv1 "sigs.k8s.io/cluster-api/api/bootstrap/kubeadm/v1beta2"
 	controlplanev1 "sigs.k8s.io/cluster-api/api/controlplane/kubeadm/v1beta2"
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
+	runtimehooksv1 "sigs.k8s.io/cluster-api/api/runtime/hooks/v1alpha1"
 	"sigs.k8s.io/cluster-api/controllers/external"
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal/etcd"
+	"sigs.k8s.io/cluster-api/internal/hooks"
 	"sigs.k8s.io/cluster-api/util/collections"
 	"sigs.k8s.io/cluster-api/util/conditions"
 	"sigs.k8s.io/cluster-api/util/failuredomains"
@@ -185,6 +187,26 @@ func (c *ControlPlane) MachineWithDeleteAnnotation(machines collections.Machines
 	return annotatedMachines
 }
 
+// MachinesToCompleteTriggerInPlaceUpdate returns Machines for which we have to complete triggering
+// the in-place update. This can become necessary if triggering the in-place update fails after
+// we added UpdateInProgressAnnotation and before we marked the UpdateMachine hook as pending.
+func (c *ControlPlane) MachinesToCompleteTriggerInPlaceUpdate() collections.Machines {
+	return c.Machines.Filter(func(machine *clusterv1.Machine) bool {
+		_, ok := machine.Annotations[clusterv1.UpdateInProgressAnnotation]
+		return ok && !hooks.IsPending(runtimehooksv1.UpdateMachine, machine)
+	})
+}
+
+// MachinesToCompleteInPlaceUpdate returns Machines that still have to complete their in-place update.
+func (c *ControlPlane) MachinesToCompleteInPlaceUpdate() collections.Machines {
+	return c.Machines.Filter(func(machine *clusterv1.Machine) bool {
+		// Note: Checking both annotations here to make this slightly more robust.
+		//       Theoretically only checking for IsPending would have been enough.
+		_, ok := machine.Annotations[clusterv1.UpdateInProgressAnnotation]
+		return ok || hooks.IsPending(runtimehooksv1.UpdateMachine, machine)
+	})
+}
+
 // FailureDomainWithMostMachines returns the fd with most machines in it and at least one eligible machine in it.
 // Note: if there are eligibleMachines machines in failure domain that do not exist anymore, cleaning up those failure domains takes precedence.
 func (c *ControlPlane) FailureDomainWithMostMachines(ctx context.Context, eligibleMachines collections.Machines) string {

controlplane/kubeadm/internal/control_plane_test.go

@@ -27,6 +27,7 @@ import (
 
 	controlplanev1 "sigs.k8s.io/cluster-api/api/controlplane/kubeadm/v1beta2"
 	clusterv1 "sigs.k8s.io/cluster-api/api/core/v1beta2"
+	runtimev1 "sigs.k8s.io/cluster-api/api/runtime/v1beta2"
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal/etcd"
 	"sigs.k8s.io/cluster-api/util/collections"
 )
@@ -293,6 +294,160 @@ func TestHasHealthyMachineStillProvisioning(t *testing.T) {
 	})
 }
 
+func TestMachinesToCompleteTriggerInPlaceUpdate(t *testing.T) {
+	machineWithoutAnnotations := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithoutAnnotations",
+		},
+	}
+	machineWithUpdateInProgressAnnotation := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithUpdateInProgressAnnotation",
+			Annotations: map[string]string{
+				clusterv1.UpdateInProgressAnnotation: "",
+			},
+		},
+	}
+	machineWithPendingHooksAnnotation := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithPendingHooksAnnotation",
+			Annotations: map[string]string{
+				runtimev1.PendingHooksAnnotation: "UpdateMachine",
+			},
+		},
+	}
+	machineWithBothAnnotations := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithBothAnnotations",
+			Annotations: map[string]string{
+				clusterv1.UpdateInProgressAnnotation: "",
+				runtimev1.PendingHooksAnnotation:     "UpdateMachine",
+			},
+		},
+	}
+
+	tests := []struct {
+		name                         string
+		machine                      *clusterv1.Machine
+		completeTriggerInPlaceUpdate bool
+	}{
+		{
+			name:                         "machineWithoutAnnotations => false",
+			machine:                      machineWithoutAnnotations,
+			completeTriggerInPlaceUpdate: false,
+		},
+		{
+			name:                         "machineWithUpdateInProgressAnnotation => true",
+			machine:                      machineWithUpdateInProgressAnnotation,
+			completeTriggerInPlaceUpdate: true,
+		},
+		{
+			name:                         "machineWithPendingHooksAnnotation => false",
+			machine:                      machineWithPendingHooksAnnotation,
+			completeTriggerInPlaceUpdate: false,
+		},
+		{
+			name:                         "machineWithBothAnnotations => false",
+			machine:                      machineWithBothAnnotations,
+			completeTriggerInPlaceUpdate: false,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			c := ControlPlane{
+				Machines: collections.FromMachines(tt.machine),
+			}
+
+			if tt.completeTriggerInPlaceUpdate {
+				g.Expect(c.MachinesToCompleteTriggerInPlaceUpdate().Len()).To(Equal(1))
+				g.Expect(c.MachinesToCompleteTriggerInPlaceUpdate().Has(tt.machine)).To(BeTrue())
+			} else {
+				g.Expect(c.MachinesToCompleteTriggerInPlaceUpdate().Len()).To(Equal(0))
+			}
+		})
+	}
+}
+
+func TestMachinesToCompleteInPlaceUpdate(t *testing.T) {
+	machineWithoutAnnotations := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithoutAnnotations",
+		},
+	}
+	machineWithUpdateInProgressAnnotation := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithUpdateInProgressAnnotation",
+			Annotations: map[string]string{
+				clusterv1.UpdateInProgressAnnotation: "",
+			},
+		},
+	}
+	machineWithPendingHooksAnnotation := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithPendingHooksAnnotation",
+			Annotations: map[string]string{
+				runtimev1.PendingHooksAnnotation: "UpdateMachine",
+			},
+		},
+	}
+	machineWithBothAnnotations := &clusterv1.Machine{
+		ObjectMeta: metav1.ObjectMeta{
+			Name: "machineWithBothAnnotations",
+			Annotations: map[string]string{
+				clusterv1.UpdateInProgressAnnotation: "",
+				runtimev1.PendingHooksAnnotation:     "UpdateMachine",
+			},
+		},
+	}
+
+	tests := []struct {
+		name                  string
+		machine               *clusterv1.Machine
+		completeInPlaceUpdate bool
+	}{
+		{
+			name:                  "machineWithoutAnnotations => false",
+			machine:               machineWithoutAnnotations,
+			completeInPlaceUpdate: false,
+		},
+		{
+			name:                  "machineWithUpdateInProgressAnnotation => true",
+			machine:               machineWithUpdateInProgressAnnotation,
+			completeInPlaceUpdate: true,
+		},
+		{
+			name:                  "machineWithPendingHooksAnnotation => true",
+			machine:               machineWithPendingHooksAnnotation,
+			completeInPlaceUpdate: true,
+		},
+		{
+			name:                  "machineWithBothAnnotations => true",
+			machine:               machineWithBothAnnotations,
+			completeInPlaceUpdate: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			g := NewWithT(t)
+
+			c := ControlPlane{
+				Machines: collections.FromMachines(tt.machine),
+			}
+
+			if tt.completeInPlaceUpdate {
+				g.Expect(c.MachinesToCompleteInPlaceUpdate().Len()).To(Equal(1))
+				g.Expect(c.MachinesToCompleteInPlaceUpdate().Has(tt.machine)).To(BeTrue())
+			} else {
+				g.Expect(c.MachinesToCompleteInPlaceUpdate().Len()).To(Equal(0))
+			}
+		})
+	}
+}
+
 func TestStatusToLogKeyAndValues(t *testing.T) {
 	healthyMachine := &clusterv1.Machine{
 		ObjectMeta: metav1.ObjectMeta{Name: "healthy"},

controlplane/kubeadm/internal/controllers/controller.go

@@ -107,6 +107,7 @@ type KubeadmControlPlaneReconciler struct {
 	overridePreflightChecksFunc        func(ctx context.Context, controlPlane *internal.ControlPlane, excludeFor ...*clusterv1.Machine) ctrl.Result
 	overrideCanUpdateMachineFunc       func(ctx context.Context, machine *clusterv1.Machine, machineUpToDateResult internal.UpToDateResult) (bool, error)
 	overrideCanExtensionsUpdateMachine func(ctx context.Context, machine *clusterv1.Machine, machineUpToDateResult internal.UpToDateResult, extensionHandlers []string) (bool, []string, error)
+	overrideTriggerInPlaceUpdate       func(ctx context.Context, machine *clusterv1.Machine, machineUpToDateResult internal.UpToDateResult) error
 	// Note: This field is only used for unit tests that use fake client because the fake client does not properly set resourceVersion
 	//       on BootstrapConfig/InfraMachine after ssa.Patch and then ssa.RemoveManagedFieldsForLabelsAndAnnotations would fail.
 	disableRemoveManagedFieldsForLabelsAndAnnotations bool
@@ -479,12 +480,33 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, controlPl
 		return result, err
 	}
 
+	// Complete triggering in-place update if necessary, for reentrancy if triggerInPlaceUpdate failed
+	// when triggering the in-place update initially.
+	if machines := controlPlane.MachinesToCompleteTriggerInPlaceUpdate(); len(machines) > 0 {
+		_, machinesUpToDateResults := controlPlane.NotUpToDateMachines()
+		for _, m := range machines {
+			if err := r.triggerInPlaceUpdate(ctx, m, machinesUpToDateResults[m.Name]); err != nil {
+				return ctrl.Result{}, err
+			}
+		}
+		return ctrl.Result{}, nil // Note: Changes to Machines trigger another reconcile.
+	}
+
 	// Reconcile unhealthy machines by triggering deletion and requeue if it is considered safe to remediate,
 	// otherwise continue with the other KCP operations.
 	if result, err := r.reconcileUnhealthyMachines(ctx, controlPlane); err != nil || !result.IsZero() {
 		return result, err
 	}
 
+	// Wait for in-place update to complete.
+	// Note: If a Machine becomes unhealthy during in-place update reconcileUnhealthyMachines above remediates it.
+	// Note: We have to wait here even if there are no more Machines that need rollout (in-place update in
+	// progress is not counted as needs rollout).
+	if machines := controlPlane.MachinesToCompleteInPlaceUpdate(); machines.Len() > 0 {
+		log.Info("Waiting for in-place update to complete", "machines", strings.Join(machines.Names(), ", "))
+		return ctrl.Result{}, nil // Note: Changes to Machines trigger another reconcile.
+	}
+
 	// Control plane machines rollout due to configuration changes (e.g. upgrades) takes precedence over other operations.
 	machinesNeedingRollout, machinesUpToDateResults := controlPlane.MachinesNeedingRollout()
 	switch {

controlplane/kubeadm/internal/controllers/inplace.go

@@ -48,6 +48,12 @@ func (r *KubeadmControlPlaneReconciler) tryInPlaceUpdate(
 		return false, resultForAllMachines, nil
 	}
 
+	// Note: Usually canUpdateMachine is only called once for a single Machine rollout.
+	// If it returns true, the code below will mark the in-place update as in progress via
+	// UpdateInProgressAnnotation. From this point forward we are not going to call canUpdateMachine again.
+	// If it returns false, we are going to fall back to scale down which will delete the Machine.
+	// We only have to repeat the canUpdateMachine call if the write call to set UpdateInProgressAnnotation
+	// fails or if we fail to delete the Machine.
 	canUpdate, err := r.canUpdateMachine(ctx, machineToInPlaceUpdate, machineUpToDateResult)
 	if err != nil {
 		return false, ctrl.Result{}, errors.Wrapf(err, "failed to determine if Machine %s can be updated in-place", machineToInPlaceUpdate.Name)
@@ -57,6 +63,5 @@ func (r *KubeadmControlPlaneReconciler) tryInPlaceUpdate(
 		return true, ctrl.Result{}, nil
 	}
 
-	// Always fallback to scale down until triggering in-place updates is implemented.
-	return true, ctrl.Result{}, nil
+	return false, ctrl.Result{}, r.triggerInPlaceUpdate(ctx, machineToInPlaceUpdate, machineUpToDateResult)
 }

controlplane/kubeadm/internal/controllers/inplace_canupdatemachine_test.go

@@ -779,7 +779,7 @@ func Test_createRequest(t *testing.T) {
 			},
 		},
 		{
-			name:                 "Should prepare all objects for diff: desiredInfraMachine picks up changes from currentMachine via SSA dry-run",
+			name:                 "Should prepare all objects for diff: desiredInfraMachine picks up changes from currentInfraMachine via SSA dry-run",
 			currentMachine:       currentMachine,
 			currentInfraMachine:  currentInfraMachine,
 			currentKubeadmConfig: currentKubeadmConfig,
@@ -809,7 +809,7 @@ func Test_createRequest(t *testing.T) {
 			},
 		},
 		{
-			name:                 "Should prepare all objects for diff: currentKubeadmConfig & currentKubeadmConfig are prepared for diff",
+			name:                 "Should prepare all objects for diff: currentKubeadmConfig & desiredKubeadmConfig are prepared for diff",
 			currentMachine:       currentMachine,
 			currentInfraMachine:  currentInfraMachine,
 			currentKubeadmConfig: currentKubeadmConfigWithInitConfiguration,
@@ -839,7 +839,6 @@ func Test_createRequest(t *testing.T) {
 
 			// Create Machine (same as in createMachine)
 			currentMachineForPatch := tt.currentMachine.DeepCopy()
-			currentMachineForPatch.SetGroupVersionKind(clusterv1.GroupVersion.WithKind("Machine")) // Has to be set for ssa.Patch
 			g.Expect(ssa.Patch(ctx, env.Client, kcpManagerName, currentMachineForPatch)).To(Succeed())
 			t.Cleanup(func() {
 				g.Expect(env.CleanupAndWait(context.Background(), tt.currentMachine)).To(Succeed())
@@ -855,7 +854,6 @@ func Test_createRequest(t *testing.T) {
 
 			// Create KubeadmConfig (same as in createKubeadmConfig)
 			currentKubeadmConfigForPatch := tt.currentKubeadmConfig.DeepCopy()
-			currentKubeadmConfigForPatch.SetGroupVersionKind(bootstrapv1.GroupVersion.WithKind("KubeadmConfig")) // Has to be set for ssa.Patch
 			g.Expect(ssa.Patch(ctx, env.Client, kcpManagerName, currentKubeadmConfigForPatch)).To(Succeed())
 			g.Expect(ssa.RemoveManagedFieldsForLabelsAndAnnotations(ctx, env.Client, env.GetAPIReader(), currentKubeadmConfigForPatch, kcpManagerName)).To(Succeed())
 			t.Cleanup(func() {

controlplane/kubeadm/internal/controllers/inplace_test.go

@@ -37,14 +37,15 @@ func Test_tryInPlaceUpdate(t *testing.T) {
 	}
 
 	tests := []struct {
-		name                       string
-		preflightChecksFunc        func(ctx context.Context, controlPlane *internal.ControlPlane, excludeFor ...*clusterv1.Machine) ctrl.Result
-		canUpdateMachineFunc       func(ctx context.Context, machine *clusterv1.Machine, machineUpToDateResult internal.UpToDateResult) (bool, error)
-		wantCanUpdateMachineCalled bool
-		wantFallbackToScaleDown    bool
-		wantError                  bool
-		wantErrorMessage           string
-		wantRes                    ctrl.Result
+		name                           string
+		preflightChecksFunc            func(ctx context.Context, controlPlane *internal.ControlPlane, excludeFor ...*clusterv1.Machine) ctrl.Result
+		canUpdateMachineFunc           func(ctx context.Context, machine *clusterv1.Machine, machineUpToDateResult internal.UpToDateResult) (bool, error)
+		wantCanUpdateMachineCalled     bool
+		wantTriggerInPlaceUpdateCalled bool
+		wantFallbackToScaleDown        bool
+		wantError                      bool
+		wantErrorMessage               string
+		wantRes                        ctrl.Result
 	}{
 		{
 			name: "Requeue if preflight checks for all Machines failed",
@@ -94,16 +95,17 @@ func Test_tryInPlaceUpdate(t *testing.T) {
 			canUpdateMachineFunc: func(_ context.Context, _ *clusterv1.Machine, _ internal.UpToDateResult) (bool, error) {
 				return true, nil
 			},
-			wantCanUpdateMachineCalled: true,
-			// TODO(in-place): Will be modified once tryInPlaceUpdate triggers in-place updates.
-			wantFallbackToScaleDown: true,
+			wantCanUpdateMachineCalled:     true,
+			wantTriggerInPlaceUpdateCalled: true,
+			wantFallbackToScaleDown:        false,
 		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			g := NewWithT(t)
 
 			var canUpdateMachineCalled bool
+			var triggerInPlaceUpdateCalled bool
 			r := &KubeadmControlPlaneReconciler{
 				overridePreflightChecksFunc: func(ctx context.Context, controlPlane *internal.ControlPlane, excludeFor ...*clusterv1.Machine) ctrl.Result {
 					return tt.preflightChecksFunc(ctx, controlPlane, excludeFor...)
@@ -112,6 +114,10 @@ func Test_tryInPlaceUpdate(t *testing.T) {
 					canUpdateMachineCalled = true
 					return tt.canUpdateMachineFunc(ctx, machine, machineUpToDateResult)
 				},
+				overrideTriggerInPlaceUpdate: func(_ context.Context, _ *clusterv1.Machine, _ internal.UpToDateResult) error {
+					triggerInPlaceUpdateCalled = true
+					return nil
+				},
 			}
 
 			fallbackToScaleDown, res, err := r.tryInPlaceUpdate(ctx, nil, machineToInPlaceUpdate, internal.UpToDateResult{})
@@ -125,6 +131,7 @@ func Test_tryInPlaceUpdate(t *testing.T) {
 			g.Expect(fallbackToScaleDown).To(Equal(tt.wantFallbackToScaleDown))
 
 			g.Expect(canUpdateMachineCalled).To(Equal(tt.wantCanUpdateMachineCalled), "canUpdateMachineCalled: actual: %t expected: %t", canUpdateMachineCalled, tt.wantCanUpdateMachineCalled)
+			g.Expect(triggerInPlaceUpdateCalled).To(Equal(tt.wantTriggerInPlaceUpdateCalled), "triggerInPlaceUpdateCalled: actual: %t expected: %t", triggerInPlaceUpdateCalled, tt.wantTriggerInPlaceUpdateCalled)
 		})
 	}
 }