reconcile rosaMachinePool version

Author: muraee

config/crd/bases/controlplane.cluster.x-k8s.io_rosacontrolplanes.yaml

@@ -362,8 +362,13 @@ spec:
                   AKS, EKS, GKE, etc.
                 type: boolean
               failureMessage:
-                description: ErrorMessage indicates that there is a terminal problem
-                  reconciling the state, and will be set to a descriptive error message.
+                description: "FailureMessage will be set in the event that there is
+                  a terminal problem reconciling the state and will be set to a descriptive
+                  error message. \n This field should not be set for transitive errors
+                  that a controller faces that are expected to be fixed automatically
+                  over time (like service outages), but instead indicate that something
+                  is fundamentally wrong with the spec or the configuration of the
+                  controller, and that manual intervention is required."
                 type: string
               id:
                 description: ID is the cluster ID given by ROSA.

config/crd/bases/infrastructure.cluster.x-k8s.io_rosamachinepools.yaml

@@ -94,6 +94,11 @@ spec:
                 type: array
               subnet:
                 type: string
+              version:
+                description: Version specifies the penshift version of the nodes associated
+                  with this machinepool. ROSAControlPlane version is used if not set.
+                pattern: ^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$
+                type: string
             required:
             - nodePoolName
             type: object
@@ -146,6 +151,15 @@ spec:
                   - type
                   type: object
                 type: array
+              failureMessage:
+                description: "FailureMessage will be set in the event that there is
+                  a terminal problem reconciling the state and will be set to a descriptive
+                  error message. \n This field should not be set for transitive errors
+                  that a controller faces that are expected to be fixed automatically
+                  over time (like service outages), but instead indicate that something
+                  is fundamentally wrong with the spec or the configuration of the
+                  controller, and that manual intervention is required."
+                type: string
               id:
                 description: ID is the ID given by ROSA.
                 type: string

controlplane/rosa/api/v1beta2/rosacontrolplane_types.go

@@ -473,8 +473,15 @@ type RosaControlPlaneStatus struct {
 	// Ready denotes that the ROSAControlPlane API Server is ready to receive requests.
 	// +kubebuilder:default=false
 	Ready bool `json:"ready"`
-	// ErrorMessage indicates that there is a terminal problem reconciling the
-	// state, and will be set to a descriptive error message.
+	// FailureMessage will be set in the event that there is a terminal problem
+	// reconciling the state and will be set to a descriptive error message.
+	//
+	// This field should not be set for transitive errors that a controller
+	// faces that are expected to be fixed automatically over
+	// time (like service outages), but instead indicate that something is
+	// fundamentally wrong with the spec or the configuration of
+	// the controller, and that manual intervention is required.
+	//
 	// +optional
 	FailureMessage *string `json:"failureMessage,omitempty"`
 	// Conditions specifies the cpnditions for the managed control plane

controlplane/rosa/controllers/rosacontrolplane_controller.go

@@ -191,13 +191,16 @@ func (r *ROSAControlPlaneReconciler) reconcileNormal(ctx context.Context, rosaSc
 	}
 	defer rosaClient.Close()
 
-	isValid, err := validateControlPlaneSpec(rosaClient, rosaScope)
+	failureMessage, err := validateControlPlaneSpec(rosaClient, rosaScope)
 	if err != nil {
 		return ctrl.Result{}, fmt.Errorf("failed to validate ROSAControlPlane.spec: %w", err)
 	}
-	if !isValid {
+	if failureMessage != nil {
+		rosaScope.ControlPlane.Status.FailureMessage = failureMessage
 		// dont' requeue because input is invalid and manual intervention is needed.
 		return ctrl.Result{}, nil
+	} else {
+		rosaScope.ControlPlane.Status.FailureMessage = nil
 	}
 
 	cluster, err := rosaClient.GetCluster()
@@ -270,7 +273,7 @@ func (r *ROSAControlPlaneReconciler) reconcileNormal(ctx context.Context, rosaSc
 		DisableUserWorkloadMonitoring(true).
 		Version(
 			cmv1.NewVersion().
-				ID(fmt.Sprintf("openshift-v%s", rosaScope.ControlPlane.Spec.Version)).
+				ID(rosa.VersionID(rosaScope.ControlPlane.Spec.Version)).
 				ChannelGroup("stable"),
 		).
 		ExpirationTimestamp(time.Now().Add(1 * time.Hour)).
@@ -411,7 +414,7 @@ func (r *ROSAControlPlaneReconciler) reconcileDelete(ctx context.Context, rosaSc
 
 func (r *ROSAControlPlaneReconciler) reconcileClusterVersion(rosaScope *scope.ROSAControlPlaneScope, rosaClient *rosa.RosaClient, cluster *cmv1.Cluster) error {
 	version := rosaScope.ControlPlane.Spec.Version
-	if version == cluster.Version().RawID() {
+	if version == rosa.RawVersionID(cluster.Version()) {
 		conditions.MarkFalse(rosaScope.ControlPlane, rosacontrolplanev1.ROSAControlPlaneUpgradingCondition, "upgraded", clusterv1.ConditionSeverityInfo, "")
 		return nil
 	}
@@ -560,24 +563,20 @@ func (r *ROSAControlPlaneReconciler) reconcileClusterAdminPassword(ctx context.C
 	return password, nil
 }
 
-func validateControlPlaneSpec(rosaClient *rosa.RosaClient, rosaScope *scope.ROSAControlPlaneScope) (bool, error) {
-	// reset previous message.
-	rosaScope.ControlPlane.Status.FailureMessage = nil
-
+func validateControlPlaneSpec(rosaClient *rosa.RosaClient, rosaScope *scope.ROSAControlPlaneScope) (*string, error) {
 	version := rosaScope.ControlPlane.Spec.Version
 	isSupported, err := rosaClient.IsVersionSupported(version)
 	if err != nil {
-		return false, err
+		return nil, fmt.Errorf("failed to verify if version is supported: %w", err)
 	}
 
 	if !isSupported {
 		message := fmt.Sprintf("version %s is not supported", version)
-		rosaScope.ControlPlane.Status.FailureMessage = &message
-		return false, nil
+		return &message, nil
 	}
 
 	// TODO: add more input validations
-	return true, nil
+	return nil, nil
 }
 
 func (r *ROSAControlPlaneReconciler) rosaClusterToROSAControlPlane(log *logger.Logger) handler.MapFunc {

exp/api/v1beta2/conditions_consts.go

@@ -104,8 +104,10 @@ const (
 )
 
 const (
-	// RosaMachinePoolReadyCondition condition reports on the successful reconciliation of rosa control plane.
+	// RosaMachinePoolReadyCondition condition reports on the successful reconciliation of rosa machinepool.
 	RosaMachinePoolReadyCondition clusterv1.ConditionType = "RosaMchinePoolReady"
+	// RosaMachinePoolUpgradingCondition condition reports whether ROSAMachinePool is upgrading or not.
+	RosaMachinePoolUpgradingCondition clusterv1.ConditionType = "RosaMchinePoolUpgrading"
 	// WaitingForRosaControlPlaneReason used when the machine pool is waiting for
 	// ROSA control plane infrastructure to be ready before proceeding.
 	WaitingForRosaControlPlaneReason = "WaitingForRosaControlPlane"

exp/api/v1beta2/rosamachinepool_types.go

@@ -33,6 +33,13 @@ type RosaMachinePoolSpec struct {
 	// +kubebuilder:validation:Pattern:=`^[a-z]([-a-z0-9]*[a-z0-9])?$`
 	NodePoolName string `json:"nodePoolName"`
 
+	// Version specifies the penshift version of the nodes associated with this machinepool.
+	// ROSAControlPlane version is used if not set.
+	//
+	// +optional
+	// +kubebuilder:validation:Pattern:=`^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d*)(?:-((?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*)(?:\.(?:0|[1-9]\d*|\d*[a-zA-Z-][0-9a-zA-Z-]*))*))?(?:\+([0-9a-zA-Z-]+(?:\.[0-9a-zA-Z-]+)*))?$`
+	Version string `json:"version,omitempty"`
+
 	// AvailabilityZone is an optinal field specifying the availability zone where instances of this machine pool should run
 	// For Multi-AZ clusters, you can create a machine pool in a Single-AZ of your choice.
 	// +optional
@@ -83,14 +90,23 @@ type RosaMachinePoolStatus struct {
 	// the cluster
 	// +kubebuilder:default=false
 	Ready bool `json:"ready"`
-
 	// Replicas is the most recently observed number of replicas.
 	// +optional
 	Replicas int32 `json:"replicas"`
-
 	// Conditions defines current service state of the managed machine pool
 	// +optional
 	Conditions clusterv1.Conditions `json:"conditions,omitempty"`
+	// FailureMessage will be set in the event that there is a terminal problem
+	// reconciling the state and will be set to a descriptive error message.
+	//
+	// This field should not be set for transitive errors that a controller
+	// faces that are expected to be fixed automatically over
+	// time (like service outages), but instead indicate that something is
+	// fundamentally wrong with the spec or the configuration of
+	// the controller, and that manual intervention is required.
+	//
+	// +optional
+	FailureMessage *string `json:"failureMessage,omitempty"`
 
 	// ID is the ID given by ROSA.
 	ID string `json:"id,omitempty"`

exp/api/v1beta2/zz_generated.deepcopy.go

@@ -5,8 +5,10 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/blang/semver"
 	cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"
 	"github.com/pkg/errors"
+	corev1 "k8s.io/api/core/v1"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/client-go/tools/record"
@@ -179,6 +181,18 @@ func (r *ROSAMachinePoolReconciler) reconcileNormal(ctx context.Context,
 	}
 	defer rosaClient.Close()
 
+	failureMessage, err := validateMachinePoolSpec(machinePoolScope)
+	if err != nil {
+		return ctrl.Result{}, fmt.Errorf("failed to validate ROSAMachinePool.spec: %w", err)
+	}
+	if failureMessage != nil {
+		machinePoolScope.RosaMachinePool.Status.FailureMessage = failureMessage
+		// dont' requeue because input is invalid and manual intervention is needed.
+		return ctrl.Result{}, nil
+	} else {
+		machinePoolScope.RosaMachinePool.Status.FailureMessage = nil
+	}
+
 	rosaMachinePool := machinePoolScope.RosaMachinePool
 	machinePool := machinePoolScope.MachinePool
 	controlPlane := machinePoolScope.ControlPlane
@@ -194,6 +208,10 @@ func (r *ROSAMachinePoolReconciler) reconcileNormal(ctx context.Context,
 			conditions.MarkTrue(rosaMachinePool, expinfrav1.RosaMachinePoolReadyCondition)
 			rosaMachinePool.Status.Ready = true
 
+			if err := r.reconcileMachinePoolVersion(machinePoolScope, rosaClient, createdNodePool); err != nil {
+				return ctrl.Result{}, err
+			}
+
 			return ctrl.Result{}, nil
 		}
 
@@ -231,6 +249,9 @@ func (r *ROSAMachinePoolReconciler) reconcileNormal(ctx context.Context,
 	}
 
 	npBuilder.AWSNodePool(cmv1.NewAWSNodePool().InstanceType(rosaMachinePool.Spec.InstanceType))
+	if rosaMachinePool.Spec.Version != "" {
+		npBuilder.Version(cmv1.NewVersion().ID(rosa.VersionID(rosaMachinePool.Spec.Version)))
+	}
 
 	nodePoolSpec, err := npBuilder.Build()
 	if err != nil {
@@ -274,6 +295,69 @@ func (r *ROSAMachinePoolReconciler) reconcileDelete(
 	return nil
 }
 
+func (r *ROSAMachinePoolReconciler) reconcileMachinePoolVersion(machinePoolScope *scope.RosaMachinePoolScope, rosaClient *rosa.RosaClient, nodePool *cmv1.NodePool) error {
+	version := machinePoolScope.RosaMachinePool.Spec.Version
+	if version == "" {
+		version = machinePoolScope.ControlPlane.Spec.Version
+	}
+
+	if version == rosa.RawVersionID(nodePool.Version()) {
+		conditions.MarkFalse(machinePoolScope.RosaMachinePool, expinfrav1.RosaMachinePoolUpgradingCondition, "upgraded", clusterv1.ConditionSeverityInfo, "")
+		return nil
+	}
+
+	clusterID := *machinePoolScope.ControlPlane.Status.ID
+	scheduledUpgrade, err := rosaClient.CheckNodePoolExistingScheduledUpgrade(clusterID, nodePool)
+	if err != nil {
+		return fmt.Errorf("failed to get existing scheduled upgrades: %w", err)
+	}
+
+	if scheduledUpgrade == nil {
+		scheduledUpgrade, err = rosaClient.ScheduleNodePoolUpgrade(clusterID, nodePool, version, time.Now())
+		if err != nil {
+			return fmt.Errorf("failed to schedule nodePool upgrade to version %s: %w", version, err)
+		}
+	}
+
+	condition := &clusterv1.Condition{
+		Type:    expinfrav1.RosaMachinePoolUpgradingCondition,
+		Status:  corev1.ConditionTrue,
+		Reason:  string(scheduledUpgrade.State().Value()),
+		Message: fmt.Sprintf("Upgrading to version %s", scheduledUpgrade.Version()),
+	}
+	conditions.Set(machinePoolScope.RosaMachinePool, condition)
+
+	// if nodePool is already upgrading to another version we need to wait until the current upgrade is finished, return an error to requeue and try later.
+	if scheduledUpgrade.Version() != version {
+		return fmt.Errorf("there is already a %s upgrade to version %s", scheduledUpgrade.State().Value(), scheduledUpgrade.Version())
+	}
+
+	return nil
+}
+
+func validateMachinePoolSpec(machinePoolScope *scope.RosaMachinePoolScope) (*string, error) {
+	if machinePoolScope.RosaMachinePool.Spec.Version == "" {
+		return nil, nil
+	}
+
+	version, err := semver.Parse(machinePoolScope.RosaMachinePool.Spec.Version)
+	if err != nil {
+		return nil, fmt.Errorf("failed to parse MachinePool version: %w", err)
+	}
+	minSupportedVersion, maxSupportedVersion, err := rosa.MachinePoolSupportedVersionsRange(machinePoolScope.ControlPlane.Spec.Version)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get supported machinePool versions range: %w", err)
+	}
+
+	if version.GT(*maxSupportedVersion) || version.LT(*minSupportedVersion) {
+		message := fmt.Sprintf("version %s is not supported, should be in the range: >= %s and <= %s", version, minSupportedVersion, maxSupportedVersion)
+		return &message, nil
+	}
+
+	// TODO: add more input validations
+	return nil, nil
+}
+
 func rosaControlPlaneToRosaMachinePoolMapFunc(c client.Client, gvk schema.GroupVersionKind, log logger.Wrapper) handler.MapFunc {
 	return func(ctx context.Context, o client.Object) []reconcile.Request {
 		rosaControlPlane, ok := o.(*rosacontrolplanev1.ROSAControlPlane)

exp/controllers/rosamachinepool_controller.go

@@ -1,6 +1,10 @@
 package rosa
 
-import cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"
+import (
+	"time"
+
+	cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"
+)
 
 // CreateNodePool adds a new node pool to the cluster.
 func (c *RosaClient) CreateNodePool(clusterID string, nodePool *cmv1.NodePool) (*cmv1.NodePool, error) {
@@ -70,3 +74,73 @@ func (c *RosaClient) DeleteNodePool(clusterID string, nodePoolID string) error {
 	}
 	return nil
 }
+
+// CheckNodePoolExistingScheduledUpgrade checks and returns the current upgrade schedule for the nodePool if any.
+func (c *RosaClient) CheckNodePoolExistingScheduledUpgrade(clusterID string, nodePool *cmv1.NodePool) (*cmv1.NodePoolUpgradePolicy, error) {
+	upgradePolicies, err := c.getNodePoolUpgradePolicies(clusterID, nodePool.ID())
+	if err != nil {
+		return nil, err
+	}
+	for _, upgradePolicy := range upgradePolicies {
+		if upgradePolicy.UpgradeType() == cmv1.UpgradeTypeNodePool {
+			return upgradePolicy, nil
+		}
+	}
+	return nil, nil
+}
+
+// ScheduleNodePoolUpgrade schedules a new nodePool upgrade to the specified version at the specified time.
+func (c *RosaClient) ScheduleNodePoolUpgrade(clusterID string, nodePool *cmv1.NodePool, version string, nextRun time.Time) (*cmv1.NodePoolUpgradePolicy, error) {
+	// earliestNextRun is set to at least 5 min from now by the OCM API.
+	// we set it to 6 min here to account for latencty.
+	earliestNextRun := time.Now().Add(time.Minute * 6)
+	if nextRun.Before(earliestNextRun) {
+		nextRun = earliestNextRun
+	}
+
+	upgradePolicy, err := cmv1.NewNodePoolUpgradePolicy().
+		UpgradeType(cmv1.UpgradeTypeNodePool).
+		NodePoolID(nodePool.ID()).
+		ScheduleType(cmv1.ScheduleTypeManual).
+		Version(version).
+		NextRun(nextRun).
+		Build()
+	if err != nil {
+		return nil, err
+	}
+
+	response, err := c.ocm.ClustersMgmt().V1().
+		Clusters().Cluster(clusterID).
+		NodePools().
+		NodePool(nodePool.ID()).UpgradePolicies().
+		Add().Body(upgradePolicy).
+		Send()
+	if err != nil {
+		return nil, handleErr(response.Error(), err)
+	}
+
+	return response.Body(), nil
+}
+
+func (c *RosaClient) getNodePoolUpgradePolicies(clusterID string, nodePoolID string) (nodePoolUpgradePolicies []*cmv1.NodePoolUpgradePolicy, err error) {
+	collection := c.ocm.ClustersMgmt().V1().
+		Clusters().
+		Cluster(clusterID).NodePools().NodePool(nodePoolID).UpgradePolicies()
+	page := 1
+	size := 100
+	for {
+		response, err := collection.List().
+			Page(page).
+			Size(size).
+			Send()
+		if err != nil {
+			return nil, handleErr(response.Error(), err)
+		}
+		nodePoolUpgradePolicies = append(nodePoolUpgradePolicies, response.Items().Slice()...)
+		if response.Size() < size {
+			break
+		}
+		page++
+	}
+	return
+}