migrate elalsticquota/capacityscheduling to controller-runtime

Author: Huang-Wei

pkg/capacityscheduling/capacity_scheduling.go

@@ -29,6 +29,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/client-go/informers"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	corelisters "k8s.io/client-go/listers/core/v1"
 	policylisters "k8s.io/client-go/listers/policy/v1"
 	"k8s.io/client-go/tools/cache"
@@ -39,23 +40,29 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler/framework/preemption"
 	"k8s.io/kubernetes/pkg/scheduler/metrics"
 	schedutil "k8s.io/kubernetes/pkg/scheduler/util"
+	ctrlruntimecache "sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling"
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling/v1alpha1"
-	"sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"
-	schedinformer "sigs.k8s.io/scheduler-plugins/pkg/generated/informers/externalversions"
-	externalv1alpha1 "sigs.k8s.io/scheduler-plugins/pkg/generated/listers/scheduling/v1alpha1"
 	"sigs.k8s.io/scheduler-plugins/pkg/util"
 )
 
+var scheme = runtime.NewScheme()
+
+func init() {
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(v1alpha1.AddToScheme(scheme))
+}
+
 // CapacityScheduling is a plugin that implements the mechanism of capacity scheduling.
 type CapacityScheduling struct {
 	sync.RWMutex
-	fh                 framework.Handle
-	podLister          corelisters.PodLister
-	pdbLister          policylisters.PodDisruptionBudgetLister
-	elasticQuotaLister externalv1alpha1.ElasticQuotaLister
-	elasticQuotaInfos  ElasticQuotaInfos
+	fh                framework.Handle
+	podLister         corelisters.PodLister
+	pdbLister         policylisters.PodDisruptionBudgetLister
+	client            client.Client
+	elasticQuotaInfos ElasticQuotaInfos
 }
 
 // PreFilterState computed at PreFilter and used at PostFilter or Reserve.
@@ -119,42 +126,43 @@ func New(obj runtime.Object, handle framework.Handle) (framework.Plugin, error)
 		pdbLister:         getPDBLister(handle.SharedInformerFactory()),
 	}
 
-	client, err := versioned.NewForConfig(handle.KubeConfig())
+	client, err := client.New(handle.KubeConfig(), client.Options{Scheme: scheme})
 	if err != nil {
 		return nil, err
 	}
 
-	schedSharedInformerFactory := schedinformer.NewSharedInformerFactory(client, 0)
-	c.elasticQuotaLister = schedSharedInformerFactory.Scheduling().V1alpha1().ElasticQuotas().Lister()
-	elasticQuotaInformer := schedSharedInformerFactory.Scheduling().V1alpha1().ElasticQuotas().Informer()
-	elasticQuotaInformer.AddEventHandler(
-		cache.FilteringResourceEventHandler{
-			FilterFunc: func(obj interface{}) bool {
-				switch t := obj.(type) {
-				case *v1alpha1.ElasticQuota:
+	c.client = client
+	dynamicCache, err := ctrlruntimecache.New(handle.KubeConfig(), ctrlruntimecache.Options{Scheme: scheme})
+	if err != nil {
+		return nil, err
+	}
+	// TODO: pass in context.
+	elasticQuotaInformer, err := dynamicCache.GetInformer(context.Background(), &v1alpha1.ElasticQuota{})
+	if err != nil {
+		return nil, err
+	}
+	elasticQuotaInformer.AddEventHandler(cache.FilteringResourceEventHandler{
+		FilterFunc: func(obj interface{}) bool {
+			switch t := obj.(type) {
+			case *v1alpha1.ElasticQuota:
+				return true
+			case cache.DeletedFinalStateUnknown:
+				if _, ok := t.Obj.(*v1alpha1.ElasticQuota); ok {
 					return true
-				case cache.DeletedFinalStateUnknown:
-					if _, ok := t.Obj.(*v1alpha1.ElasticQuota); ok {
-						return true
-					}
-					utilruntime.HandleError(fmt.Errorf("cannot convert to *v1alpha1.ElasticQuota: %v", obj))
-					return false
-				default:
-					utilruntime.HandleError(fmt.Errorf("unable to handle object in %T", obj))
-					return false
 				}
-			},
-			Handler: cache.ResourceEventHandlerFuncs{
-				AddFunc:    c.addElasticQuota,
-				UpdateFunc: c.updateElasticQuota,
-				DeleteFunc: c.deleteElasticQuota,
-			},
-		})
-
-	schedSharedInformerFactory.Start(nil)
-	if !cache.WaitForCacheSync(nil, elasticQuotaInformer.HasSynced) {
-		return nil, fmt.Errorf("timed out waiting for caches to sync %v", Name)
-	}
+				utilruntime.HandleError(fmt.Errorf("cannot convert to *v1alpha1.ElasticQuota: %v", obj))
+				return false
+			default:
+				utilruntime.HandleError(fmt.Errorf("unable to handle object in %T", obj))
+				return false
+			}
+		},
+		Handler: cache.ResourceEventHandlerFuncs{
+			AddFunc:    c.addElasticQuota,
+			UpdateFunc: c.updateElasticQuota,
+			DeleteFunc: c.deleteElasticQuota,
+		},
+	})
 
 	podInformer := handle.SharedInformerFactory().Core().V1().Pods().Informer()
 	podInformer.AddEventHandler(
@@ -688,12 +696,13 @@ func (c *CapacityScheduling) addPod(obj interface{}) {
 	elasticQuotaInfo := c.elasticQuotaInfos[pod.Namespace]
 	// If elasticQuotaInfo is nil, try to list ElasticQuotas through elasticQuotaLister
 	if elasticQuotaInfo == nil {
-		eqs, err := c.elasticQuotaLister.ElasticQuotas(pod.Namespace).List(labels.NewSelector())
-		if err != nil {
+		var eqList v1alpha1.ElasticQuotaList
+		if err := c.client.List(context.Background(), &eqList, client.InNamespace(pod.Namespace)); err != nil {
 			klog.ErrorS(err, "Failed to get elasticQuota", "elasticQuota", pod.Namespace)
 			return
 		}
 
+		eqs := eqList.Items
 		// If the length of elasticQuotas is 0, return.
 		if len(eqs) == 0 {
 			return

test/integration/capacity_scheduling_test.go

@@ -22,21 +22,20 @@ import (
 	"time"
 
 	v1 "k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/errors"
+	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/util/wait"
 	"k8s.io/client-go/kubernetes"
-	"k8s.io/klog/v2"
 	"k8s.io/kubernetes/pkg/scheduler"
 	schedapi "k8s.io/kubernetes/pkg/scheduler/apis/config"
 	fwkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
+	"sigs.k8s.io/controller-runtime/pkg/client"
 
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling"
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling/v1alpha1"
 	"sigs.k8s.io/scheduler-plugins/pkg/capacityscheduling"
-	"sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"
 	"sigs.k8s.io/scheduler-plugins/test/util"
 )
 
@@ -45,7 +44,7 @@ func TestCapacityScheduling(t *testing.T) {
 	testCtx.Ctx, testCtx.CancelFn = context.WithCancel(context.Background())
 
 	cs := kubernetes.NewForConfigOrDie(globalKubeConfig)
-	extClient := versioned.NewForConfigOrDie(globalKubeConfig)
+	extClient := util.NewClientOrDie(globalKubeConfig)
 	testCtx.ClientSet = cs
 	testCtx.KubeConfig = globalKubeConfig
 
@@ -557,21 +556,17 @@ func TestCapacityScheduling(t *testing.T) {
 	}
 }
 
-func createElasticQuotas(ctx context.Context, client versioned.Interface, elasticQuotas []*v1alpha1.ElasticQuota) error {
+func createElasticQuotas(ctx context.Context, client client.Client, elasticQuotas []*v1alpha1.ElasticQuota) error {
 	for _, eq := range elasticQuotas {
-		_, err := client.SchedulingV1alpha1().ElasticQuotas(eq.Namespace).Create(ctx, eq, metav1.CreateOptions{})
-		if err != nil && !errors.IsAlreadyExists(err) {
+		if err := client.Create(ctx, eq); err != nil && !apierrors.IsAlreadyExists(err) {
 			return err
 		}
 	}
 	return nil
 }
 
-func cleanupElasticQuotas(ctx context.Context, client versioned.Interface, elasticQuotas []*v1alpha1.ElasticQuota) {
+func cleanupElasticQuotas(ctx context.Context, client client.Client, elasticQuotas []*v1alpha1.ElasticQuota) {
 	for _, eq := range elasticQuotas {
-		err := client.SchedulingV1alpha1().ElasticQuotas(eq.Namespace).Delete(ctx, eq.Name, metav1.DeleteOptions{})
-		if err != nil {
-			klog.ErrorS(err, "Failed to clean up ElasticQuota", "elasticQuota", klog.KObj(eq))
-		}
+		_ = client.Delete(ctx, eq)
 	}
 }

test/integration/elasticquota_controller_test.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/wait"
 	quota "k8s.io/apiserver/pkg/quota/v1"
@@ -34,23 +35,22 @@ import (
 	"k8s.io/kubernetes/pkg/scheduler"
 	fwkruntime "k8s.io/kubernetes/pkg/scheduler/framework/runtime"
 	st "k8s.io/kubernetes/pkg/scheduler/testing"
-
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/manager"
 
 	"sigs.k8s.io/scheduler-plugins/apis/scheduling"
 	schedv1alpha1 "sigs.k8s.io/scheduler-plugins/apis/scheduling/v1alpha1"
 	"sigs.k8s.io/scheduler-plugins/pkg/capacityscheduling"
 	"sigs.k8s.io/scheduler-plugins/pkg/controllers"
-	"sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"
+	"sigs.k8s.io/scheduler-plugins/test/util"
 )
 
 func TestElasticController(t *testing.T) {
 	testCtx := &testContext{}
 	testCtx.Ctx, testCtx.CancelFn = context.WithCancel(context.Background())
 
 	cs := kubernetes.NewForConfigOrDie(globalKubeConfig)
-	extClient := versioned.NewForConfigOrDie(globalKubeConfig)
+	extClient := util.NewClientOrDie(globalKubeConfig)
 	testCtx.ClientSet = cs
 	testCtx.KubeConfig = globalKubeConfig
 
@@ -311,8 +311,8 @@ func TestElasticController(t *testing.T) {
 
 			if err := wait.Poll(time.Millisecond*200, 10*time.Second, func() (bool, error) {
 				for _, v := range tt.used {
-					eq, err := extClient.SchedulingV1alpha1().ElasticQuotas(v.Namespace).Get(testCtx.Ctx, v.Name, metav1.GetOptions{})
-					if err != nil {
+					var eq schedv1alpha1.ElasticQuota
+					if err := extClient.Get(testCtx.Ctx, types.NamespacedName{Namespace: v.Namespace, Name: v.Name}, &eq); err != nil {
 						// This could be a connection error so we want to retry.
 						klog.ErrorS(err, "Failed to obtain the elasticQuota clientSet")
 						return false, err
@@ -345,8 +345,8 @@ func TestElasticController(t *testing.T) {
 
 			if err := wait.Poll(time.Millisecond*200, 10*time.Second, func() (bool, error) {
 				for _, v := range tt.want {
-					eq, err := extClient.SchedulingV1alpha1().ElasticQuotas(v.Namespace).Get(testCtx.Ctx, v.Name, metav1.GetOptions{})
-					if err != nil {
+					var eq schedv1alpha1.ElasticQuota
+					if err := extClient.Get(testCtx.Ctx, types.NamespacedName{Namespace: v.Namespace, Name: v.Name}, &eq); err != nil {
 						// This could be a connection error so we want to retry.
 						klog.ErrorS(err, "Failed to obtain the elasticQuota clientSet")
 						return false, err