Merge pull request #665 from Huang-Wei/elasticquota-migrate-controller-runtime

migrate elasticquota/capacityscheduling to controller-runtime

Author: k8s-ci-robot

Makefile

@@ -112,7 +112,7 @@ update-vendor:
 	hack/update-vendor.sh
 
 .PHONY: unit-test
-unit-test:
+unit-test: install-envtest
 	hack/unit-test.sh
 
 .PHONY: install-envtest

cmd/scheduler/main_test.go

@@ -20,19 +20,19 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"net/http"
-	"net/http/httptest"
 	"os"
 	"path/filepath"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/spf13/pflag"
 
+	clientcmdapi "k8s.io/client-go/tools/clientcmd/api"
 	"k8s.io/kubernetes/cmd/kube-scheduler/app"
 	"k8s.io/kubernetes/cmd/kube-scheduler/app/options"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config"
 	"k8s.io/kubernetes/pkg/scheduler/apis/config/testing/defaults"
+	"sigs.k8s.io/controller-runtime/pkg/envtest"
 
 	"sigs.k8s.io/scheduler-plugins/pkg/capacityscheduling"
 	"sigs.k8s.io/scheduler-plugins/pkg/coscheduling"
@@ -48,42 +48,47 @@ import (
 )
 
 func TestSetup(t *testing.T) {
+	testEnv := &envtest.Environment{
+		CRDDirectoryPaths: []string{
+			filepath.Join("..", "..", "manifests", "crds"),
+		},
+	}
+
+	// start envtest cluster
+	cfg, err := testEnv.Start()
+	defer testEnv.Stop()
+	if err != nil {
+		panic(err)
+	}
+
 	// temp dir
 	tmpDir, err := os.MkdirTemp("", "scheduler-options")
 	if err != nil {
 		t.Fatal(err)
 	}
 	defer os.RemoveAll(tmpDir)
 
-	// https server
-	server := httptest.NewTLSServer(http.HandlerFunc(func(w http.ResponseWriter, req *http.Request) {
-		w.Header().Set("Content-Type", "application/json")
-		w.WriteHeader(http.StatusOK)
-		w.Write([]byte(`{"metadata": {"name": "test"}}`))
-	}))
-	defer server.Close()
+	clusters := make(map[string]*clientcmdapi.Cluster)
+	clusters["default-cluster"] = &clientcmdapi.Cluster{
+		Server:                   cfg.Host,
+		CertificateAuthorityData: cfg.CAData,
+	}
+	// https://github.com/kubernetes-sigs/controller-runtime/blob/v0.16.3/examples/scratch-env/main.go
+	user, err := testEnv.ControlPlane.AddUser(envtest.User{
+		Name:   "envtest-admin",
+		Groups: []string{"system:masters"},
+	}, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	kubeConfig, err := user.KubeConfig()
+	if err != nil {
+		t.Fatalf("unable to create kubeconfig: %v", err)
+	}
 
 	configKubeconfig := filepath.Join(tmpDir, "config.kubeconfig")
-	if err := os.WriteFile(configKubeconfig, []byte(fmt.Sprintf(`
-apiVersion: v1
-kind: Config
-clusters:
-- cluster:
-    insecure-skip-tls-verify: true
-    server: %s
-  name: default
-contexts:
-- context:
-    cluster: default
-    user: default
-  name: default
-current-context: default
-users:
-- name: default
-  user:
-    username: config
-`, server.URL)), os.FileMode(0600)); err != nil {
-		t.Fatal(err)
+	if err := os.WriteFile(configKubeconfig, kubeConfig, os.FileMode(0600)); err != nil {
+		t.Fatalf("unable to create kubeconfig file: %v", err)
 	}
 
 	// PodState plugin config

hack/unit-test.sh

@@ -21,6 +21,10 @@ set -o pipefail
 SCRIPT_ROOT=$(dirname "${BASH_SOURCE}")/..
 source "${SCRIPT_ROOT}/hack/lib/init.sh"
 
+kube::log::status "Configuring envtest"
+TEMP_DIR=${TMPDIR-/tmp}
+source "${TEMP_DIR}/setup-envtest"
+
 # TODO: make args customizable.
 go test -mod=vendor \
   sigs.k8s.io/scheduler-plugins/cmd/... \

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