add: Hypershift Functionality

Author: VanillaSpoon

controllers/appwrapper_controller.go

@@ -18,10 +18,11 @@ package controllers
 
 import (
 	"context"
-	"fmt"
 	"strings"
 	"time"
 
+	ocmsdk "github.com/openshift-online/ocm-sdk-go"
+
 	"github.com/project-codeflare/instascale/pkg/config"
 	arbv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
 	corev1 "k8s.io/api/core/v1"
@@ -37,15 +38,19 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
+type MachineType string
+
 // AppWrapperReconciler reconciles a AppWrapper object
 type AppWrapperReconciler struct {
 	client.Client
-	Scheme         *runtime.Scheme
-	Config         config.InstaScaleConfiguration
-	kubeClient     *kubernetes.Clientset
-	ocmClusterID   string
-	ocmToken       string
-	useMachineSets bool
+	Scheme        *runtime.Scheme
+	Config        config.InstaScaleConfiguration
+	kubeClient    *kubernetes.Clientset
+	ocmClusterID  string
+	ocmToken      string
+	ocmConnection *ocmsdk.Connection
+	MachineType   MachineType
+	machineCheck  bool
 }
 
 var (
@@ -54,9 +59,12 @@ var (
 )
 
 const (
-	namespaceToList = "openshift-machine-api"
-	minResyncPeriod = 10 * time.Minute
-	finalizerName   = "instascale.codeflare.dev/finalizer"
+	namespaceToList                    = "openshift-machine-api"
+	minResyncPeriod                    = 10 * time.Minute
+	finalizerName                      = "instascale.codeflare.dev/finalizer"
+	MachineTypeMachineSet  MachineType = "MachineSet"
+	MachineTypeMachinePool MachineType = "MachinePool"
+	MachineTypeNodePool    MachineType = "NodePool"
 )
 
 // +kubebuilder:rbac:groups=workload.codeflare.dev,resources=appwrappers,verbs=get;list;watch;create;update;patch;delete
@@ -81,14 +89,17 @@ const (
 // - https://pkg.go.dev/sigs.k8s.io/[email protected]/pkg/reconcile
 func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
 	_ = log.FromContext(ctx)
-	// todo: Move the getOCMClusterID call out of reconcile loop.
+	// todo: Move the setMachineType call out of reconcile loop.
 	// Only reason we are calling it here is that the client is not able to make
 	// calls until it is started, so SetupWithManager is not working.
-	if !r.useMachineSets && r.ocmClusterID == "" {
-		if err := r.getOCMClusterID(ctx); err != nil {
-			return ctrl.Result{Requeue: true, RequeueAfter: timeFiveSeconds}, err
+	if r.machineCheck == false && r.MachineType != MachineTypeMachineSet {
+		if err := r.setMachineType(ctx); err != nil {
+			return ctrl.Result{}, err
 		}
 	}
+
+	r.machineCheck = true
+
 	var appwrapper arbv1.AppWrapper
 
 	if err := r.Get(ctx, req.NamespacedName, &appwrapper); err != nil {
@@ -125,8 +136,14 @@ func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 	if status == "Pending" && containsInsufficientCondition(allconditions) {
 		demandPerInstanceType := r.discoverInstanceTypes(ctx, &appwrapper)
 		if ocmSecretRef := r.Config.OCMSecretRef; ocmSecretRef != nil {
-			return r.scaleMachinePool(ctx, &appwrapper, demandPerInstanceType)
+			switch r.MachineType {
+			case MachineTypeNodePool:
+				return r.scaleNodePool(ctx, &appwrapper, demandPerInstanceType)
+			case MachineTypeMachinePool:
+				return r.scaleMachinePool(ctx, &appwrapper, demandPerInstanceType)
+			}
 		} else {
+			// use MachineSets
 			switch strings.ToLower(r.Config.MachineSetsStrategy) {
 			case "reuse":
 				return r.reconcileReuseMachineSet(ctx, &appwrapper, demandPerInstanceType)
@@ -146,7 +163,8 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 	} else {
 		deletionMessage = "deleted"
 	}
-	if r.useMachineSets {
+	switch r.MachineType {
+	case MachineTypeMachineSet:
 		switch strings.ToLower(r.Config.MachineSetsStrategy) {
 		case "reuse":
 			matchedAw := r.findExactMatch(ctx, appwrapper)
@@ -158,6 +176,9 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 					"newAppWrapper", matchedAw,
 				)
 				if err := r.swapNodeLabels(ctx, appwrapper, matchedAw); err != nil {
+					logger.Error(err, "Error swapping node labels for AppWrapper",
+						"appwrapper", appwrapper,
+					)
 					return err
 				}
 			} else {
@@ -167,6 +188,9 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 					"deletionMessage", deletionMessage,
 				)
 				if err := r.annotateToDeleteMachine(ctx, appwrapper); err != nil {
+					logger.Error(err, "Error annotating to delete machine for AppWrapper",
+						"appwrapper", appwrapper,
+					)
 					return err
 				}
 			}
@@ -177,19 +201,53 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 				"deletionMessage", deletionMessage,
 			)
 			if err := r.deleteMachineSet(ctx, appwrapper); err != nil {
+				logger.Error(err, "Error deleting MachineSet for AppWrapper",
+					"appwrapper", appwrapper)
 				return err
 			}
 		}
-	} else {
+	case MachineTypeNodePool:
+		logger.Info(
+			"AppWrapper deleted, scaling down nodepool",
+			"appWrapper", appwrapper,
+			"deletionMessage", deletionMessage,
+		)
+		if _, err := r.deleteNodePool(ctx, appwrapper); err != nil {
+			logger.Error(err, "Error deleting NodePool for AppWrapper",
+				"appwrapper", appwrapper)
+			return err
+		}
+
+	case MachineTypeMachinePool:
 		logger.Info(
 			"AppWrapper deleted, scaling down machine pool",
 			"appWrapper", appwrapper,
 			"deletionMessage", deletionMessage,
 		)
 		if _, err := r.deleteMachinePool(ctx, appwrapper); err != nil {
+			logger.Error(err, "Error deleting MachinePool for AppWrapper",
+				"appwrapper", appwrapper)
+			return err
+		}
+	}
+	return nil
+}
+
+func (r *AppWrapperReconciler) setMachineType(ctx context.Context) error {
+	logger := ctrl.LoggerFrom(ctx)
+	if r.ocmClusterID == "" {
+		if err := r.getOCMClusterID(ctx); err != nil {
 			return err
 		}
 	}
+	hypershiftEnabled, err := r.checkHypershiftEnabled(ctx)
+	if err != nil {
+		logger.Error(err, "error checking if hypershift is enabled")
+		return err
+	}
+	if hypershiftEnabled {
+		r.MachineType = MachineTypeNodePool
+	}
 	return nil
 }
 
@@ -206,20 +264,19 @@ func (r *AppWrapperReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Ma
 	}
 
 	maxScaleNodesAllowed = int(r.Config.MaxScaleoutAllowed)
-	r.useMachineSets = true
+	r.MachineType = MachineTypeMachineSet // default to MachineSet
 	if ocmSecretRef := r.Config.OCMSecretRef; ocmSecretRef != nil {
-		r.useMachineSets = false
 		if ocmSecret, err := r.getOCMSecret(ctx, ocmSecretRef); err != nil {
-			return fmt.Errorf("error reading OCM Secret from ref %q: %w", ocmSecretRef, err)
+			logger.Error(err, "error reading OCM Secret from ref",
+				"ref", ocmSecretRef)
+			return err
 		} else if token := ocmSecret.Data["token"]; len(token) > 0 {
 			r.ocmToken = string(token)
+			r.MachineType = MachineTypeMachinePool
 		} else {
-			return fmt.Errorf("token is missing from OCM Secret %q", ocmSecretRef)
-		}
-		if ok, err := r.machinePoolExists(); err != nil {
+			logger.Error(err, "token is missing from OCM Secret",
+				"ref", ocmSecretRef)
 			return err
-		} else if ok {
-			logger.Info("Using machine pools for cluster auto-scaling")
 		}
 	}
 
@@ -235,12 +292,8 @@ func (r *AppWrapperReconciler) getOCMSecret(ctx context.Context, secretRef *core
 func (r *AppWrapperReconciler) discoverInstanceTypes(ctx context.Context, aw *arbv1.AppWrapper) map[string]int {
 	logger := ctrl.LoggerFrom(ctx)
 	demandMapPerInstanceType := make(map[string]int)
-	var instanceRequired []string
-	for k, v := range aw.Labels {
-		if k == "orderedinstance" {
-			instanceRequired = strings.Split(v, "_")
-		}
-	}
+
+	instanceRequired := getInstanceRequired(aw.Labels)
 
 	if len(instanceRequired) < 1 {
 		logger.Info(

controllers/machinepools.go

@@ -2,45 +2,14 @@ package controllers
 
 import (
 	"context"
-	"fmt"
 	"strings"
 
-	ocmsdk "github.com/openshift-online/ocm-sdk-go"
 	cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"
-	configv1 "github.com/openshift/api/config/v1"
 	arbv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
 
-	"k8s.io/apimachinery/pkg/types"
 	ctrl "sigs.k8s.io/controller-runtime"
 )
 
-func (r *AppWrapperReconciler) createOCMConnection() (*ocmsdk.Connection, error) {
-	logger, err := ocmsdk.NewGoLoggerBuilder().
-		Debug(false).
-		Build()
-	if err != nil {
-		return nil, fmt.Errorf("can't build logger: %v", err)
-	}
-
-	connection, err := ocmsdk.NewConnectionBuilder().
-		Logger(logger).
-		Tokens(r.ocmToken).
-		Build()
-	if err != nil {
-		return nil, fmt.Errorf("can't build connection: %v", err)
-	}
-
-	return connection, nil
-}
-
-func hasAwLabel(machinePool *cmv1.MachinePool, aw *arbv1.AppWrapper) bool {
-	value, ok := machinePool.Labels()[aw.Name]
-	if ok && value == aw.Name {
-		return true
-	}
-	return false
-}
-
 func (r *AppWrapperReconciler) scaleMachinePool(ctx context.Context, aw *arbv1.AppWrapper, demandPerInstanceType map[string]int) (ctrl.Result, error) {
 	logger := ctrl.LoggerFrom(ctx)
 	connection, err := r.createOCMConnection()
@@ -61,7 +30,7 @@ func (r *AppWrapperReconciler) scaleMachinePool(ctx context.Context, aw *arbv1.A
 
 		numberOfMachines := 0
 		response.Items().Each(func(machinePool *cmv1.MachinePool) bool {
-			if machinePool.InstanceType() == userRequestedInstanceType && hasAwLabel(machinePool, aw) {
+			if machinePool.InstanceType() == userRequestedInstanceType && hasAwLabel(machinePool.Labels(), aw) {
 				numberOfMachines = machinePool.Replicas()
 				return false
 			}
@@ -133,52 +102,3 @@ func (r *AppWrapperReconciler) deleteMachinePool(ctx context.Context, aw *arbv1.
 	})
 	return ctrl.Result{Requeue: false}, nil
 }
-
-func (r *AppWrapperReconciler) machinePoolExists() (bool, error) {
-	connection, err := r.createOCMConnection()
-	if err != nil {
-		return false, fmt.Errorf("error creating OCM connection: %w", err)
-	}
-	defer connection.Close()
-
-	machinePools := connection.ClustersMgmt().V1().Clusters().Cluster(r.ocmClusterID).MachinePools()
-	return machinePools != nil, nil
-}
-
-// getOCMClusterID determines the internal clusterID to be used for OCM API calls
-func (r *AppWrapperReconciler) getOCMClusterID(ctx context.Context) error {
-	logger := ctrl.LoggerFrom(ctx)
-	cv := &configv1.ClusterVersion{}
-	err := r.Get(ctx, types.NamespacedName{Name: "version"}, cv)
-	if err != nil {
-		return fmt.Errorf("can't get clusterversion: %v", err)
-	}
-
-	internalClusterID := string(cv.Spec.ClusterID)
-
-	connection, err := r.createOCMConnection()
-	if err != nil {
-		logger.Error(err, "Error creating OCM connection")
-	}
-	defer connection.Close()
-
-	// Get the client for the resource that manages the collection of clusters:
-	collection := connection.ClustersMgmt().V1().Clusters()
-
-	response, err := collection.List().Search(fmt.Sprintf("external_id = '%s'", internalClusterID)).Size(1).Page(1).SendContext(ctx)
-	if err != nil {
-		logger.Error(err, "Error getting cluster id")
-	}
-
-	response.Items().Each(func(cluster *cmv1.Cluster) bool {
-		r.ocmClusterID = cluster.ID()
-		logger.Info(
-			"Cluster Info",
-			"clusterId", cluster.ID(),
-			"clusterName", cluster.Name(),
-			"clusterState", cluster.State(),
-		)
-		return true
-	})
-	return nil
-}