move capi machine k8s version update logic from node upgradeto CPupgr…

…ade controller

Signed-off-by: Rahul Ganesh <[email protected]>

controllers/controlplaneupgrade_controller.go

@@ -26,6 +26,7 @@ import (
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	kerrors "k8s.io/apimachinery/pkg/util/errors"
+	clusterv1 "sigs.k8s.io/cluster-api/api/v1beta1"
 	"sigs.k8s.io/cluster-api/util/patch"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -214,6 +215,19 @@ func (r *ControlPlaneUpgradeReconciler) updateStatus(ctx context.Context, log lo
 			return fmt.Errorf("getting node upgrader for machine %s: %v", machineRef.Name, err)
 		}
 		if nodeUpgrade.Status.Completed {
+			machine, err := getCapiMachine(ctx, r.client, nodeUpgrade)
+			if err != nil {
+				return err
+			}
+			machinePatchHelper, err := patch.NewHelper(machine, r.client)
+			if err != nil {
+				return err
+			}
+			log.Info("Updating K8s version in  machine", "Machine", machine.Name)
+			machine.Spec.Version = &nodeUpgrade.Spec.KubernetesVersion
+			if err := machinePatchHelper.Patch(ctx, machine); err != nil {
+				return fmt.Errorf("updating status for machine %s: %v", machine.Name, err)
+			}
 			nodesUpgradeCompleted++
 			nodesUpgradeRequired--
 		}
@@ -224,3 +238,11 @@ func (r *ControlPlaneUpgradeReconciler) updateStatus(ctx context.Context, log lo
 	cpUpgrade.Status.Ready = nodesUpgradeRequired == 0
 	return nil
 }
+
+func getCapiMachine(ctx context.Context, client client.Client, nodeUpgrade *anywherev1.NodeUpgrade) (*clusterv1.Machine, error) {
+	machine := &clusterv1.Machine{}
+	if err := client.Get(ctx, GetNamespacedNameType(nodeUpgrade.Spec.Machine.Name, nodeUpgrade.Spec.Machine.Namespace), machine); err != nil {
+		return nil, fmt.Errorf("getting machine %s: %v", nodeUpgrade.Spec.Machine.Name, err)
+	}
+	return machine, nil
+}

controllers/controlplaneupgrade_controller_test.go

@@ -3,6 +3,7 @@ package controllers_test
 import (
 	"context"
 	"fmt"
+	"strings"
 	"testing"
 
 	. "github.com/onsi/gomega"
@@ -191,6 +192,33 @@ func TestCPUpgradeObjectDoesNotExist(t *testing.T) {
 	g.Expect(err).To(MatchError("controlplaneupgrades.anywhere.eks.amazonaws.com \"cp-upgrade-request\" not found"))
 }
 
+func TestCPUpgradeReconcileUpdateCapiMachineVersion(t *testing.T) {
+	g := NewWithT(t)
+	ctx := context.Background()
+
+	cluster, machines, nodes, cpUpgrade, nodeUpgrades := getObjectsForCPUpgradeTest()
+	for i := range nodeUpgrades {
+		nodeUpgrades[i].Name = fmt.Sprintf("%s-node-upgrader", machines[i].Name)
+		nodeUpgrades[i].Status = anywherev1.NodeUpgradeStatus{
+			Completed: true,
+		}
+	}
+	objs := []runtime.Object{cluster, machines[0], machines[1], nodes[0], nodes[1], cpUpgrade, nodeUpgrades[0], nodeUpgrades[1]}
+	nodeUpgrades[0].Status.Completed = true
+	client := fake.NewClientBuilder().WithRuntimeObjects(objs...).Build()
+	r := controllers.NewControlPlaneUpgradeReconciler(client)
+
+	req := cpUpgradeRequest(cpUpgrade)
+	_, err := r.Reconcile(ctx, req)
+	g.Expect(err).ToNot(HaveOccurred())
+	machine := &clusterv1.Machine{}
+	err = client.Get(ctx, types.NamespacedName{Name: nodeUpgrades[0].Spec.Machine.Name, Namespace: "eksa-system"}, machine)
+	g.Expect(err).ToNot(HaveOccurred())
+	if !strings.Contains(*machine.Spec.Version, "v1.28.3-eks-1-28-9") {
+		t.Fatalf("unexpected k8s version in capi machine: %s", *machine.Spec.Version)
+	}
+}
+
 func getObjectsForCPUpgradeTest() (*clusterv1.Cluster, []*clusterv1.Machine, []*corev1.Node, *anywherev1.ControlPlaneUpgrade, []*anywherev1.NodeUpgrade) {
 	cluster := generateCluster()
 	node1 := generateNode()

controllers/nodeupgrade_controller.go

@@ -6,6 +6,7 @@ import (
 	"time"
 
 	"github.com/go-logr/logr"
+	"github.com/pkg/errors"
 	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -25,10 +26,8 @@ import (
 )
 
 const (
-	// TODO(in-place): Get this image from the bundle instead of using the hardcoded one.
-	defaultUpgraderImage = "public.ecr.aws/t2p4l7v3/upgrader:eksdbase"
-	controlPlaneLabel    = "node-role.kubernetes.io/control-plane"
-	podDNEMessage        = "Upgrader pod does not exist"
+	controlPlaneLabel = "node-role.kubernetes.io/control-plane"
+	podDNEMessage     = "Upgrader pod does not exist"
 
 	// nodeUpgradeFinalizerName is the finalizer added to NodeUpgrade objects to handle deletion.
 	nodeUpgradeFinalizerName = "nodeupgrades.anywhere.eks.amazonaws.com/finalizer"
@@ -66,6 +65,7 @@ func (r *NodeUpgradeReconciler) SetupWithManager(mgr ctrl.Manager) error {
 //+kubebuilder:rbac:groups=anywhere.eks.amazonaws.com,resources=nodeupgrades/status,verbs=get;update;patch
 //+kubebuilder:rbac:groups=anywhere.eks.amazonaws.com,resources=nodeupgrades/finalizers,verbs=update
 //+kubebuilder:rbac:groups="",resources=pods,verbs=get;list;watch;create;delete
+//+kubebuilder:rbac:groups="",resources=configmaps,verbs=get;list;watch
 //+kubebuilder:rbac:groups="cluster.x-k8s.io",resources=machines,verbs=list;watch;get;patch;update
 
 // Reconcile reconciles a NodeUpgrade object.
@@ -100,18 +100,13 @@ func (r *NodeUpgradeReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	}
 
 	// Initialize the patch helper
-	nodeUpgradePatchHelper, err := patch.NewHelper(nodeUpgrade, r.client)
-	if err != nil {
-		return ctrl.Result{}, err
-	}
-
-	capiMachinePatchHelper, err := patch.NewHelper(machineToBeUpgraded, r.client)
+	patchHelper, err := patch.NewHelper(nodeUpgrade, r.client)
 	if err != nil {
 		return ctrl.Result{}, err
 	}
 
 	defer func() {
-		err := r.updateStatus(ctx, log, rClient, nodeUpgrade, machineToBeUpgraded)
+		err := r.updateStatus(ctx, log, rClient, nodeUpgrade, machineToBeUpgraded.Status.NodeRef.Name)
 		if err != nil {
 			reterr = kerrors.NewAggregate([]error{reterr, err})
 		}
@@ -127,13 +122,10 @@ func (r *NodeUpgradeReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 		if reterr == nil {
 			patchOpts = append(patchOpts, patch.WithStatusObservedGeneration{})
 		}
-		if err := patchNodeUpgrade(ctx, nodeUpgradePatchHelper, *nodeUpgrade, patchOpts...); err != nil {
+		if err := patchNodeUpgrade(ctx, patchHelper, *nodeUpgrade, patchOpts...); err != nil {
 			reterr = kerrors.NewAggregate([]error{reterr, err})
 		}
 
-		if err := capiMachinePatchHelper.Patch(ctx, machineToBeUpgraded); err != nil {
-			reterr = kerrors.NewAggregate([]error{reterr, err})
-		}
 		// Only requeue if we are not already re-queueing and the NodeUpgrade ready condition is false.
 		// We do this to be able to update the status continuously until the NodeUpgrade becomes ready,
 		// since there might be changes in state of the world that don't trigger reconciliation requests
@@ -176,14 +168,26 @@ func (r *NodeUpgradeReconciler) reconcile(ctx context.Context, log logr.Logger,
 		return ctrl.Result{}, nil
 	}
 
+	configMap := &corev1.ConfigMap{}
+	if err := r.client.Get(ctx, types.NamespacedName{Name: constants.UpgraderConfigMapName, Namespace: constants.EksaSystemNamespace}, configMap); err != nil {
+		return ctrl.Result{}, err
+	}
+	if configMap.Data == nil {
+		return ctrl.Result{}, errors.New("upgrader config map is empty")
+	}
+	upgraderImage, ok := configMap.Data[nodeUpgrade.Spec.KubernetesVersion]
+	if !ok {
+		return ctrl.Result{}, fmt.Errorf("upgrader image corresponding to EKS Distro version %s not found in the config map", nodeUpgrade.Spec.KubernetesVersion)
+	}
+
 	if isControlPlane(node) {
 		if nodeUpgrade.Spec.FirstNodeToBeUpgraded {
-			upgraderPod = upgrader.UpgradeFirstControlPlanePod(node.Name, defaultUpgraderImage, nodeUpgrade.Spec.KubernetesVersion, *nodeUpgrade.Spec.EtcdVersion)
+			upgraderPod = upgrader.UpgradeFirstControlPlanePod(node.Name, upgraderImage, nodeUpgrade.Spec.KubernetesVersion, *nodeUpgrade.Spec.EtcdVersion)
 		} else {
-			upgraderPod = upgrader.UpgradeSecondaryControlPlanePod(node.Name, defaultUpgraderImage)
+			upgraderPod = upgrader.UpgradeSecondaryControlPlanePod(node.Name, upgraderImage)
 		}
 	} else {
-		upgraderPod = upgrader.UpgradeWorkerPod(node.Name, defaultUpgraderImage)
+		upgraderPod = upgrader.UpgradeWorkerPod(node.Name, upgraderImage)
 	}
 
 	if err := remoteClient.Create(ctx, upgraderPod); err != nil {
@@ -238,7 +242,7 @@ func (r *NodeUpgradeReconciler) reconcileDelete(ctx context.Context, log logr.Lo
 	return ctrl.Result{}, nil
 }
 
-func (r *NodeUpgradeReconciler) updateStatus(ctx context.Context, log logr.Logger, remoteClient client.Client, nodeUpgrade *anywherev1.NodeUpgrade, machine *clusterv1.Machine) error {
+func (r *NodeUpgradeReconciler) updateStatus(ctx context.Context, log logr.Logger, remoteClient client.Client, nodeUpgrade *anywherev1.NodeUpgrade, nodeName string) error {
 	// When NodeUpgrade is fully deleted, we do not need to update the status. Without this check
 	// the subsequent patch operations would fail if the status is updated after it is fully deleted.
 	if !nodeUpgrade.DeletionTimestamp.IsZero() && len(nodeUpgrade.GetFinalizers()) == 0 {
@@ -248,7 +252,7 @@ func (r *NodeUpgradeReconciler) updateStatus(ctx context.Context, log logr.Logge
 
 	log.Info("Updating NodeUpgrade status")
 
-	pod, err := getUpgraderPod(ctx, remoteClient, machine.Status.NodeRef.Name)
+	pod, err := getUpgraderPod(ctx, remoteClient, nodeName)
 	if err != nil {
 		if apierrors.IsNotFound(err) {
 			markAllConditionsFalse(nodeUpgrade, podDNEMessage, clusterv1.ConditionSeverityInfo)
@@ -261,10 +265,6 @@ func (r *NodeUpgradeReconciler) updateStatus(ctx context.Context, log logr.Logge
 	conditions.MarkTrue(nodeUpgrade, anywherev1.UpgraderPodCreated)
 	updateComponentsConditions(pod, nodeUpgrade)
 
-	if nodeUpgrade.Status.Completed {
-		log.Info("Updating machine status", "Machine", machine.Name)
-		machine.Spec.Version = &nodeUpgrade.Spec.KubernetesVersion
-	}
 	// Always update the readyCondition by summarizing the state of other conditions.
 	conditions.SetSummary(nodeUpgrade,
 		conditions.WithConditions(