in-place propagation for KCP

Author: Yuvaraj Kakaraparthi

controlplane/kubeadm/internal/control_plane.go

@@ -49,8 +49,8 @@ type ControlPlane struct {
 
 	// TODO: we should see if we can combine these with the Machine objects so we don't have all these separate lookups
 	// See discussion on https://github.com/kubernetes-sigs/cluster-api/pull/3405
-	kubeadmConfigs map[string]*bootstrapv1.KubeadmConfig
-	infraResources map[string]*unstructured.Unstructured
+	KubeadmConfigs map[string]*bootstrapv1.KubeadmConfig
+	InfraResources map[string]*unstructured.Unstructured
 }
 
 // NewControlPlane returns an instantiated ControlPlane.
@@ -77,8 +77,8 @@ func NewControlPlane(ctx context.Context, client client.Client, cluster *cluster
 		Cluster:              cluster,
 		Machines:             ownedMachines,
 		machinesPatchHelpers: patchHelpers,
-		kubeadmConfigs:       kubeadmConfigs,
-		infraResources:       infraObjects,
+		KubeadmConfigs:       kubeadmConfigs,
+		InfraResources:       infraObjects,
 		reconciliationTime:   metav1.Now(),
 	}, nil
 }
@@ -158,7 +158,7 @@ func (c *ControlPlane) HasDeletingMachine() bool {
 
 // GetKubeadmConfig returns the KubeadmConfig of a given machine.
 func (c *ControlPlane) GetKubeadmConfig(machineName string) (*bootstrapv1.KubeadmConfig, bool) {
-	kubeadmConfig, ok := c.kubeadmConfigs[machineName]
+	kubeadmConfig, ok := c.KubeadmConfigs[machineName]
 	return kubeadmConfig, ok
 }
 
@@ -169,15 +169,15 @@ func (c *ControlPlane) MachinesNeedingRollout() collections.Machines {
 
 	// Return machines if they are scheduled for rollout or if with an outdated configuration.
 	return machines.Filter(
-		NeedsRollout(&c.reconciliationTime, c.KCP.Spec.RolloutAfter, c.KCP.Spec.RolloutBefore, c.infraResources, c.kubeadmConfigs, c.KCP),
+		NeedsRollout(&c.reconciliationTime, c.KCP.Spec.RolloutAfter, c.KCP.Spec.RolloutBefore, c.InfraResources, c.KubeadmConfigs, c.KCP),
 	)
 }
 
 // UpToDateMachines returns the machines that are up to date with the control
 // plane's configuration and therefore do not require rollout.
 func (c *ControlPlane) UpToDateMachines() collections.Machines {
 	return c.Machines.Filter(
-		collections.Not(NeedsRollout(&c.reconciliationTime, c.KCP.Spec.RolloutAfter, c.KCP.Spec.RolloutBefore, c.infraResources, c.kubeadmConfigs, c.KCP)),
+		collections.Not(NeedsRollout(&c.reconciliationTime, c.KCP.Spec.RolloutAfter, c.KCP.Spec.RolloutBefore, c.InfraResources, c.KubeadmConfigs, c.KCP)),
 	)
 }
 
@@ -258,3 +258,8 @@ func (c *ControlPlane) PatchMachines(ctx context.Context) error {
 	}
 	return kerrors.NewAggregate(errList)
 }
+
+// SetPatchHelpers updates the patch helpers.
+func (c *ControlPlane) SetPatchHelpers(patchHelpers map[string]*patch.Helper) {
+	c.machinesPatchHelpers = patchHelpers
+}

controlplane/kubeadm/internal/controllers/controller.go

@@ -44,7 +44,8 @@ import (
 	"sigs.k8s.io/cluster-api/controlplane/kubeadm/internal"
 	expv1 "sigs.k8s.io/cluster-api/exp/api/v1beta1"
 	"sigs.k8s.io/cluster-api/feature"
-	"sigs.k8s.io/cluster-api/internal/labels"
+	"sigs.k8s.io/cluster-api/internal/contract"
+	"sigs.k8s.io/cluster-api/internal/util/ssa"
 	"sigs.k8s.io/cluster-api/util"
 	"sigs.k8s.io/cluster-api/util/annotations"
 	"sigs.k8s.io/cluster-api/util/collections"
@@ -55,6 +56,8 @@ import (
 	"sigs.k8s.io/cluster-api/util/version"
 )
 
+const kcpManagerName = "capi-kubeadmcontrolplane"
+
 // +kubebuilder:rbac:groups=core,resources=events,verbs=get;list;watch;create;patch
 // +kubebuilder:rbac:groups=core,resources=secrets,verbs=get;list;watch;create;update;patch
 // +kubebuilder:rbac:groups=infrastructure.cluster.x-k8s.io;bootstrap.cluster.x-k8s.io;controlplane.cluster.x-k8s.io,resources=*,verbs=get;list;watch;create;update;patch;delete
@@ -77,6 +80,12 @@ type KubeadmControlPlaneReconciler struct {
 
 	managementCluster         internal.ManagementCluster
 	managementClusterUncached internal.ManagementCluster
+
+	// disableInPlacePropagation should only be used for tests. This is used to skip
+	// some parts of the controller that need SSA as the current test setup does not
+	// support SSA. This flag should be dropped after all affected tests are migrated
+	// to envtest.
+	disableInPlacePropagation bool
 }
 
 func (r *KubeadmControlPlaneReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager, options controller.Options) error {
@@ -331,25 +340,16 @@ func (r *KubeadmControlPlaneReconciler) reconcile(ctx context.Context, cluster *
 		return ctrl.Result{}, err
 	}
 
+	if !r.disableInPlacePropagation {
+		if err := r.syncMachines(ctx, controlPlane); err != nil {
+			return ctrl.Result{}, errors.Wrap(err, "failed to sync Machines")
+		}
+	}
+
 	// Aggregate the operational state of all the machines; while aggregating we are adding the
 	// source ref (reason@machine/name) so the problem can be easily tracked down to its source machine.
 	conditions.SetAggregate(controlPlane.KCP, controlplanev1.MachinesReadyCondition, ownedMachines.ConditionGetters(), conditions.AddSourceRef(), conditions.WithStepCounterIf(false))
 
-	// Ensure all required labels exist on the controlled Machines.
-	// This logic is needed to add the `cluster.x-k8s.io/control-plane-name` label to Machines
-	// which were created before the `cluster.x-k8s.io/control-plane-name` label was introduced
-	// or if a user manually removed the label.
-	// NOTE: Changes will be applied to the Machines in reconcileControlPlaneConditions.
-	// NOTE: cluster.x-k8s.io/control-plane is already set at this stage (it is used when reading controlPlane.Machines).
-	for i := range controlPlane.Machines {
-		machine := controlPlane.Machines[i]
-		// Note: MustEqualValue and MustFormatValue is used here as the label value can be a hash if the control plane
-		// name is longer than 63 characters.
-		if value, ok := machine.Labels[clusterv1.MachineControlPlaneNameLabel]; !ok || !labels.MustEqualValue(kcp.Name, value) {
-			machine.Labels[clusterv1.MachineControlPlaneNameLabel] = labels.MustFormatValue(kcp.Name)
-		}
-	}
-
 	// Updates conditions reporting the status of static pods and the status of the etcd cluster.
 	// NOTE: Conditions reporting KCP operation progress like e.g. Resized or SpecUpToDate are inlined with the rest of the execution.
 	if result, err := r.reconcileControlPlaneConditions(ctx, controlPlane); err != nil || !result.IsZero() {
@@ -535,6 +535,86 @@ func (r *KubeadmControlPlaneReconciler) ClusterToKubeadmControlPlane(o client.Ob
 	return nil
 }
 
+// syncMachines updates Machines, InfrastructureMachines and KubeadmConfigs to propagate in-place mutable fields from KCP.
+// Note: It also cleans up managed fields of all Machines so that Machines that were
+// created/patched before (< v1.4.0) the controller adopted Server-Side-Apply (SSA) can also work with SSA.
+// Note: For InfrastructureMachines and KubeadmConfigs it also drops ownership of "metadata.labels" and
+// "metadata.annotations" from "manager" so that "capi-kubeadmcontrolplane" can own these fields and can work with SSA.
+// Otherwise, fields would be co-owned by our "old" "manager" and "capi-kubeadmcontrolplane" and then we would not be
+// able to e.g. drop labels and annotations.
+func (r *KubeadmControlPlaneReconciler) syncMachines(ctx context.Context, controlPlane *internal.ControlPlane) error {
+	patchHelpers := map[string]*patch.Helper{}
+	for machineName := range controlPlane.Machines {
+		m := controlPlane.Machines[machineName]
+		// If the machine is already being deleted, we don't need to update it.
+		if !m.DeletionTimestamp.IsZero() {
+			continue
+		}
+
+		// Cleanup managed fields of all Machines.
+		// We do this so that Machines that were created/patched before the controller adopted Server-Side-Apply (SSA)
+		// (< v1.4.0) can also work with SSA. Otherwise, fields would be co-owned by our "old" "manager" and
+		// "capi-kubeadmcontrolplane" and then we would not be able to e.g. drop labels and annotations.
+		if err := ssa.CleanUpManagedFieldsForSSAAdoption(ctx, r.Client, m, kcpManagerName); err != nil {
+			return errors.Wrapf(err, "failed to update Machine: failed to adjust the managedFields of the Machine %s", klog.KObj(m))
+		}
+		// Update Machine to propagate in-place mutable fields from KCP.
+		updatedMachine, err := r.updateMachine(ctx, m, controlPlane.KCP, controlPlane.Cluster)
+		if err != nil {
+			return errors.Wrapf(err, "failed to update Machine: %s", klog.KObj(m))
+		}
+		controlPlane.Machines[machineName] = updatedMachine
+		// Since the machine is updated, re-create the patch helper so that any subsequent
+		// Patch calls use the correct base machine object to calculate the diffs.
+		// Example: reconcileControlPlaneConditions patches the machine objects in a subsequent call
+		// and, it should use the updated machine to calculate the diff.
+		// Note: If the patchHelpers are not re-computed based on the new updated machines, subsequent
+		// Patch calls will fail because the patch will be calculated based on an outdated machine and will error
+		// because of outdated resourceVersion.
+		// TODO: This should be cleaned-up to have a more streamline way of constructing and using patchHelpers.
+		patchHelper, err := patch.NewHelper(updatedMachine, r.Client)
+		if err != nil {
+			return errors.Wrapf(err, "failed to create patch helper for Machine %s", klog.KObj(updatedMachine))
+		}
+		patchHelpers[machineName] = patchHelper
+
+		labelsAndAnnotationsManagedFieldPaths := []contract.Path{
+			{"f:metadata", "f:annotations"},
+			{"f:metadata", "f:labels"},
+		}
+		infraMachine := controlPlane.InfraResources[machineName]
+		// Cleanup managed fields of all InfrastructureMachines to drop ownership of labels and annotations
+		// from "manager". We do this so that InfrastructureMachines that are created using the Create method
+		// can also work with SSA. Otherwise, labels and annotations would be co-owned by our "old" "manager"
+		// and "capi-kubeadmcontrolplane" and then we would not be able to e.g. drop labels and annotations.
+		if err := ssa.DropManagedFields(ctx, r.Client, infraMachine, kcpManagerName, labelsAndAnnotationsManagedFieldPaths); err != nil {
+			return errors.Wrapf(err, "failed to clean up managedFields of InfrastructureMachine %s", klog.KObj(infraMachine))
+		}
+		// Update in-place mutating fields on InfrastructureMachine.
+		if err := r.updateExternalObject(ctx, infraMachine, controlPlane.KCP, controlPlane.Cluster); err != nil {
+			return errors.Wrapf(err, "failed to update InfrastructureMachine %s", klog.KObj(infraMachine))
+		}
+
+		kubeadmConfig := controlPlane.KubeadmConfigs[machineName]
+		// Note: Set the GroupVersionKind because updateExternalObject depends on it.
+		kubeadmConfig.SetGroupVersionKind(m.Spec.Bootstrap.ConfigRef.GroupVersionKind())
+		// Cleanup managed fields of all KubeadmConfigs to drop ownership of labels and annotations
+		// from "manager". We do this so that KubeadmConfigs that are created using the Create method
+		// can also work with SSA. Otherwise, labels and annotations would be co-owned by our "old" "manager"
+		// and "capi-kubeadmcontrolplane" and then we would not be able to e.g. drop labels and annotations.
+		if err := ssa.DropManagedFields(ctx, r.Client, kubeadmConfig, kcpManagerName, labelsAndAnnotationsManagedFieldPaths); err != nil {
+			return errors.Wrapf(err, "failed to clean up managedFields of KubeadmConfig %s", klog.KObj(kubeadmConfig))
+		}
+		// Update in-place mutating fields on BootstrapConfig.
+		if err := r.updateExternalObject(ctx, kubeadmConfig, controlPlane.KCP, controlPlane.Cluster); err != nil {
+			return errors.Wrapf(err, "failed to update KubeadmConfig %s", klog.KObj(kubeadmConfig))
+		}
+	}
+	// Update the patch helpers.
+	controlPlane.SetPatchHelpers(patchHelpers)
+	return nil
+}
+
 // reconcileControlPlaneConditions is responsible of reconciling conditions reporting the status of static pods and
 // the status of the etcd cluster.
 func (r *KubeadmControlPlaneReconciler) reconcileControlPlaneConditions(ctx context.Context, controlPlane *internal.ControlPlane) (ctrl.Result, error) {