add: hypershift functionality

Author: VanillaSpoon

controllers/appwrapper_controller.go

@@ -19,9 +19,7 @@ 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"
@@ -30,6 +28,8 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/klog"
+	"strings"
+	"time"
 
 	"k8s.io/apimachinery/pkg/labels"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -38,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 (
@@ -55,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,15 +88,25 @@ const (
 // For more details, check Reconcile and its Result here:
 // - 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 "machineCheck" locic 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 {
+		if r.MachineType != MachineTypeMachineSet && r.ocmClusterID == "" {
+			if err := r.getOCMClusterID(); err != nil {
+				return ctrl.Result{}, err
+			}
+		}
+		hypershiftEnabled, err := r.checkHypershiftEnabled(ctx)
+		if err != nil {
+			return ctrl.Result{}, fmt.Errorf("error checking if hypershift is enabled: %w", err)
 		}
+		if hypershiftEnabled {
+			r.MachineType = MachineTypeNodePool
+		}
+		r.machineCheck = true
 	}
 	var appwrapper arbv1.AppWrapper
 
@@ -124,8 +141,14 @@ func (r *AppWrapperReconciler) Reconcile(ctx context.Context, req ctrl.Request)
 
 	demandPerInstanceType := r.discoverInstanceTypes(&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)
@@ -142,7 +165,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)
@@ -163,7 +187,14 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 				return err
 			}
 		}
-	} else {
+
+	case MachineTypeNodePool:
+		klog.Infof("Appwrapper %s scale-down node pool: %s ", deletionMessage, appwrapper.Name)
+		if _, err := r.deleteNodePool(ctx, appwrapper); err != nil {
+			return err
+		}
+
+	case MachineTypeMachinePool:
 		klog.Infof("Appwrapper %s scale-down machine pool: %s ", deletionMessage, appwrapper.Name)
 		if _, err := r.deleteMachinePool(ctx, appwrapper); err != nil {
 			return err
@@ -174,7 +205,6 @@ func (r *AppWrapperReconciler) finalizeScalingDownMachines(ctx context.Context,
 
 // SetupWithManager sets up the controller with the Manager.
 func (r *AppWrapperReconciler) SetupWithManager(ctx context.Context, mgr ctrl.Manager) error {
-
 	restConfig := mgr.GetConfig()
 
 	var err error
@@ -184,21 +214,16 @@ 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)
 		} 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 {
-			return err
-		} else if ok {
-			klog.Info("Using machine pools for cluster auto-scaling")
-		}
 	}
 
 	return ctrl.NewControllerManagedBy(mgr).

controllers/machinepools.go

@@ -6,43 +6,13 @@ import (
 	"os"
 	"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"
 	"k8s.io/klog"
 	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) {
 	connection, err := r.createOCMConnection()
 	if err != nil {
@@ -62,7 +32,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
 			}
@@ -115,46 +85,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 {
-	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 {
-		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
-	}
-	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 {
-		klog.Errorf(`Error getting cluster id: %v`, err)
-	}
-
-	response.Items().Each(func(cluster *cmv1.Cluster) bool {
-		r.ocmClusterID = cluster.ID()
-		fmt.Printf("%s - %s - %s\n", cluster.ID(), cluster.Name(), cluster.State())
-		return true
-	})
-	return nil
-}

controllers/nodepools.go

@@ -0,0 +1,117 @@
+package controllers
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"strings"
+
+	cmv1 "github.com/openshift-online/ocm-sdk-go/clustersmgmt/v1"
+	arbv1 "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/apis/controller/v1beta1"
+
+	"k8s.io/klog"
+	ctrl "sigs.k8s.io/controller-runtime"
+)
+
+func (r *AppWrapperReconciler) scaleNodePool(ctx context.Context, aw *arbv1.AppWrapper, demandPerInstanceType map[string]int) (ctrl.Result, error) {
+	connection, err := r.createOCMConnection()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
+		return ctrl.Result{}, err
+	}
+	defer connection.Close()
+	for userRequestedInstanceType := range demandPerInstanceType {
+		replicas := demandPerInstanceType[userRequestedInstanceType]
+
+		clusterNodePools := connection.ClustersMgmt().V1().Clusters().Cluster(r.ocmClusterID).NodePools()
+
+		response, err := clusterNodePools.List().SendContext(ctx)
+		if err != nil {
+			return ctrl.Result{}, err
+		}
+
+		numberOfMachines := 0
+		response.Items().Each(func(nodePool *cmv1.NodePool) bool {
+
+			if nodePool.AWSNodePool().InstanceType() == userRequestedInstanceType && hasAwLabel(nodePool.Labels(), aw) {
+				numberOfMachines = nodePool.Replicas()
+				return false
+			}
+			return true
+		})
+
+		if numberOfMachines != replicas {
+			m := make(map[string]string)
+			m[aw.Name] = aw.Name
+			klog.Infof("The instanceRequired array: %v", userRequestedInstanceType)
+
+			nodePoolID := strings.ReplaceAll(aw.Name+"-"+userRequestedInstanceType, ".", "-")
+
+			createNodePool, err := cmv1.NewNodePool().AWSNodePool(cmv1.NewAWSNodePool().InstanceType(userRequestedInstanceType)).ID(nodePoolID).Replicas(replicas).Labels(m).Build()
+			if err != nil {
+				klog.Errorf(`Error building NodePool: %v`, err)
+			}
+			klog.Infof("Built NodePool with instance type %v and name %v", userRequestedInstanceType, createNodePool.ID())
+			response, err := clusterNodePools.Add().Body(createNodePool).SendContext(ctx)
+			if err != nil {
+				klog.Errorf(`Error creating NodePool: %v`, err)
+			}
+			klog.Infof("Created NodePool: %v", response)
+		}
+	}
+	return ctrl.Result{Requeue: false}, nil
+}
+
+func (r *AppWrapperReconciler) deleteNodePool(ctx context.Context, aw *arbv1.AppWrapper) (ctrl.Result, error) {
+	connection, err := r.createOCMConnection()
+	if err != nil {
+		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
+		return ctrl.Result{}, err
+	}
+	defer connection.Close()
+
+	nodePoolsConnection := connection.ClustersMgmt().V1().Clusters().Cluster(r.ocmClusterID).NodePools().List()
+
+	nodePoolsListResponse, _ := nodePoolsConnection.Send()
+	nodePoolsList := nodePoolsListResponse.Items()
+	nodePoolsList.Range(func(index int, item *cmv1.NodePool) bool {
+		id, _ := item.GetID()
+		if strings.Contains(id, aw.Name) {
+			targetNodePool, err := connection.ClustersMgmt().V1().Clusters().Cluster(r.ocmClusterID).NodePools().NodePool(id).Delete().SendContext(ctx)
+			if err != nil {
+				klog.Infof("Error deleting target nodepool %v", targetNodePool)
+			}
+			klog.Infof("Successfully Scaled down target nodepool %v", id)
+		}
+		return true
+	})
+	return ctrl.Result{Requeue: false}, nil
+}
+
+func (r *AppWrapperReconciler) checkHypershiftEnabled(ctx context.Context) (bool, error) {
+	connection, err := r.createOCMConnection()
+	if err != nil {
+		return false, fmt.Errorf("error creating OCM connection: %w", err)
+	}
+	defer connection.Close()
+
+	clusterResource := connection.ClustersMgmt().V1().Clusters().Cluster(r.ocmClusterID)
+
+	response, err := clusterResource.Get().SendContext(ctx)
+	if err != nil {
+		return false, fmt.Errorf("error fetching cluster details: %w", err)
+	}
+
+	body := response.Body()
+	if body == nil {
+		return false, fmt.Errorf("empty response body")
+	}
+
+	hypershiftEnabled := false
+	if body.Hypershift() != nil {
+		hypershiftEnabled = body.Hypershift().Enabled()
+	}
+
+	fmt.Printf("Hypershift enabled status: %v\n", hypershiftEnabled)
+	return hypershiftEnabled, nil
+}