add: integration with hypershift

Author: VanillaSpoon

controllers/appwrapper_controller.go

@@ -18,13 +18,13 @@ package controllers
 
 import (
 	"context"
+	"fmt"
 	"os"
 	"strconv"
 	"strings"
 	"time"
 
 	mapiclientset "github.com/openshift/client-go/machine/clientset/versioned"
-	machineinformersv1beta1 "github.com/openshift/client-go/machine/informers/externalversions"
 	"github.com/openshift/client-go/machine/listers/machine/v1beta1"
 
 	appwrapperClientSet "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/client/clientset/versioned"
@@ -33,10 +33,13 @@ import (
 	appwrapperlisters "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/client/listers/controller/v1beta1"
 
 	arbinformersFactory "github.com/project-codeflare/multi-cluster-app-dispatcher/pkg/client/informers/externalversions"
+
+	machineinformersv1beta1 "github.com/openshift/client-go/machine/informers/externalversions"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
 	"k8s.io/client-go/tools/cache"
 	"k8s.io/klog"
@@ -51,15 +54,20 @@ type AppWrapperReconciler struct {
 	Scheme             *runtime.Scheme
 	ConfigsNamespace   string
 	OcmSecretNamespace string
+	KubeClient         *kubernetes.Clientset
+	InformerFactory    informers.SharedInformerFactory
+	StopCh             chan struct{}
 }
 
 var (
-	scaledAppwrapper []string
-	reuse            = true
-	ocmClusterID     string
-	ocmToken         string
-	useMachineSets   bool
-
+	scaledAppwrapper     []string
+	scalingType          string
+	reuse                = true
+	ocmClusterID         string
+	ocmToken             string
+	useMachineSets       bool
+	useMachinePool       bool
+	useNodePool          bool
 	maxScaleNodesAllowed int
 	msLister             v1beta1.MachineSetLister
 	machineLister        v1beta1.MachineLister
@@ -92,46 +100,75 @@ 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)
 
 	var appwrapper arbv1.AppWrapper
 	if err := r.Get(ctx, req.NamespacedName, &appwrapper); err != nil {
 		if apierrors.IsNotFound(err) {
-			//ignore not-found errors, since we can get them on delete requests.
+			// ignore not-found errors, since we can get them on delete requests.
 			return ctrl.Result{}, nil
 		}
 		klog.Error(err, "unable to fetch appwrapper")
 	}
 
-	factory := machineinformersv1beta1.NewSharedInformerFactoryWithOptions(machineClient, resyncPeriod()(), machineinformersv1beta1.WithNamespace(""))
-	informer := factory.Machine().V1beta1().MachineSets().Informer()
-	msLister = factory.Machine().V1beta1().MachineSets().Lister()
-	machineLister = factory.Machine().V1beta1().Machines().Lister()
+	// This block can be moved outside the if-else condition if both the node and machineset use cases require it
 
-	// todo: Move the getOCMClusterID 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 !useMachineSets && ocmClusterID == "" {
-		getOCMClusterID(r)
+	if ocmClusterID == "" {
+		ocmManager, err := NewOCMManager()
+		if err != nil {
+			klog.Error("Failed to initialize OCMManager:", err)
+			return ctrl.Result{}, err
+		}
+		defer ocmManager.Close()
+
+		err = ocmManager.getOCMClusterID(r)
+		if err != nil {
+			klog.Error("Failed to get OCM Cluster ID:", err)
+			return ctrl.Result{}, err
+		}
 	}
 
-	stopper := make(chan struct{})
-	defer close(stopper)
-	informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
-		AddFunc:    onAdd,
-		UpdateFunc: onUpdate,
-		DeleteFunc: onDelete,
-	})
-	go informer.Run(stopper)
-	if !cache.WaitForCacheSync(stopper, informer.HasSynced) {
-		klog.Info("Wait for cache to sync")
+	if !useNodePool {
+		factory := machineinformersv1beta1.NewSharedInformerFactoryWithOptions(machineClient, resyncPeriod()(), machineinformersv1beta1.WithNamespace(""))
+		informer := factory.Machine().V1beta1().MachineSets().Informer()
+		msLister = factory.Machine().V1beta1().MachineSets().Lister()
+		machineLister = factory.Machine().V1beta1().Machines().Lister()
+
+		stopper := make(chan struct{})
+		defer close(stopper)
+		informer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc:    onAdd,
+			UpdateFunc: onUpdate,
+			DeleteFunc: onDelete,
+		})
+		go informer.Run(stopper)
+		if !cache.WaitForCacheSync(stopper, informer.HasSynced) {
+			klog.Info("Wait for cache to sync")
+		}
+		msInformerHasSynced = informer.HasSynced()
+
+	} else {
+		nodeInformer := r.InformerFactory.Core().V1().Nodes().Informer()
+		stopper := make(chan struct{})
+		defer close(stopper)
+		nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{
+			AddFunc:    onAdd,
+			UpdateFunc: onUpdate,
+			DeleteFunc: onDelete,
+		})
+		go nodeInformer.Run(stopper)
+		if !cache.WaitForCacheSync(stopper, nodeInformer.HasSynced) {
+			klog.Info("Failed to wait for node informer cache to sync")
+			return ctrl.Result{}, fmt.Errorf("failed to wait for node informer cache to sync")
+		}
+		//TODO: do we need dual sync??
+		msInformerHasSynced = nodeInformer.HasSynced()
 	}
-	//TODO: do we need dual sync??
-	msInformerHasSynced = informer.HasSynced()
+
 	addAppwrappersThatNeedScaling()
-	<-stopper
 	return ctrl.Result{}, nil
 }
 
@@ -166,13 +203,21 @@ func (r *AppWrapperReconciler) SetupWithManager(mgr ctrl.Manager) error {
 	}
 
 	useMachineSets = true
+	useMachinePool = false
+	useNodePool = false
+
 	ocmSecretExists := ocmSecretExists(r.OcmSecretNamespace)
 	if ocmSecretExists {
-		machinePoolsExists := machinePoolExists()
-
-		if machinePoolsExists {
+		nodepoolexists := nodePoolExists()
+		machinepoolexists := machinePoolExists()
+		if nodepoolexists {
+			useMachineSets = false
+			useNodePool = true
+			klog.Infof("Using node pools %v", nodepoolexists)
+		} else if machinepoolexists {
 			useMachineSets = false
-			klog.Infof("Using machine pools %v", machinePoolsExists)
+			useMachinePool = true
+			klog.Infof("Setting useMachinePools to %v", machinepoolexists)
 		} else {
 			klog.Infof("Setting useMachineSets to %v", useMachineSets)
 		}
@@ -245,7 +290,7 @@ func onAdd(obj interface{}) {
 			demandPerInstanceType := discoverInstanceTypes(aw)
 
 			if demandPerInstanceType != nil {
-				if (useMachineSets && canScaleMachineset(demandPerInstanceType)) || (!useMachineSets && canScaleMachinepool(demandPerInstanceType)) {
+				if (useMachineSets && canScaleMachineset(demandPerInstanceType)) || (!useMachineSets && (canScaleNodepool(demandPerInstanceType) || canScaleMachinepool(demandPerInstanceType))) {
 					scaleUp(aw, demandPerInstanceType)
 				} else {
 					klog.Infof("Cannot scale up replicas. The maximum allowed replicas is %v", maxScaleNodesAllowed)
@@ -312,6 +357,10 @@ func canScaleMachinepool(demandPerInstanceType map[string]int) bool {
 	return true
 }
 
+func canScaleNodepool(demandPerInstanceType map[string]int) bool {
+	return true
+}
+
 func scaleUp(aw *arbv1.AppWrapper, demandMapPerInstanceType map[string]int) {
 	if msInformerHasSynced {
 		//Assumption is made that the cluster has machineset configure that AW needs
@@ -321,6 +370,8 @@ func scaleUp(aw *arbv1.AppWrapper, demandMapPerInstanceType map[string]int) {
 
 			if useMachineSets {
 				scaleMachineSet(aw, userRequestedInstanceType, replicas)
+			} else if useNodePool {
+				scaleNodePool(aw, userRequestedInstanceType, replicas)
 			} else {
 				scaleMachinePool(aw, userRequestedInstanceType, replicas)
 			}
@@ -401,6 +452,8 @@ func onDelete(obj interface{}) {
 				klog.Infof("Appwrapper deleted scale-down machineset: %s ", aw.Name)
 				scaleDown(aw)
 			}
+		} else if useNodePool {
+			deleteNodePool(aw)
 		} else {
 			deleteMachinePool(aw)
 		}

controllers/machinepools.go

@@ -2,43 +2,20 @@ package controllers
 
 import (
 	"context"
-	"fmt"
-	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"
-	"os"
 	"strings"
 )
 
-func 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(ocmToken).
-		Build()
-	if err != nil {
-		return nil, fmt.Errorf("can't build connection: %v", err)
-	}
-
-	return connection, nil
-}
-
 func scaleMachinePool(aw *arbv1.AppWrapper, userRequestedInstanceType string, replicas int) {
-	connection, err := createOCMConnection()
+	ocmManager, err := NewOCMManager()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
-		return
+		klog.Error("Failed to initialize OCMManager:", err)
 	}
-	defer connection.Close()
+	defer ocmManager.Close()
+
+	connection := ocmManager.Connection
 
 	clusterMachinePools := connection.ClustersMgmt().V1().Clusters().Cluster(ocmClusterID).MachinePools()
 
@@ -59,12 +36,13 @@ func scaleMachinePool(aw *arbv1.AppWrapper, userRequestedInstanceType string, re
 }
 
 func deleteMachinePool(aw *arbv1.AppWrapper) {
-	connection, err := createOCMConnection()
+	ocmManager, err := NewOCMManager()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
-		return
+		klog.Error("Failed to initialize OCMManager:", err)
 	}
-	defer connection.Close()
+	defer ocmManager.Close()
+
+	connection := ocmManager.Connection
 
 	machinePoolsConnection := connection.ClustersMgmt().V1().Clusters().Cluster(ocmClusterID).MachinePools().List()
 
@@ -84,47 +62,20 @@ func deleteMachinePool(aw *arbv1.AppWrapper) {
 
 // Check if machine pools exist
 func machinePoolExists() bool {
-	connection, err := createOCMConnection()
+	ocmManager, err := NewOCMManager()
 	if err != nil {
-		fmt.Fprintf(os.Stderr, "Error creating OCM connection: %v", err)
+		klog.Error("Failed to initialize OCMManager:", err)
 	}
-	defer connection.Close()
+	defer ocmManager.Close()
+
+	connection := ocmManager.Connection
 
 	machinePools := connection.ClustersMgmt().V1().Clusters().Cluster(ocmClusterID).MachinePools()
 	klog.Infof("Machine pools: %v", machinePools)
-	return machinePools != nil
-}
-
-// getOCMClusterID determines the internal clusterID to be used for OCM API calls
-func getOCMClusterID(r *AppWrapperReconciler) error {
-	cv := &configv1.ClusterVersion{}
-	err := r.Client.Get(context.TODO(), types.NamespacedName{Name: "version"}, cv)
-	if err != nil {
-		return fmt.Errorf("can't get clusterversion: %v", err)
-	}
-
-	internalClusterID := string(cv.Spec.ClusterID)
-
-	ctx := context.Background()
-
-	connection, err := 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 {
-		ocmClusterID = cluster.ID()
-		fmt.Printf("%s - %s - %s\n", cluster.ID(), cluster.Name(), cluster.State())
+	if machinePools.List() != nil {
+		klog.Infof("machine pools are present %v", machinePools)
 		return true
-	})
-	return nil
+	}
+	return false
 }