feat: migrate podGroup controller to ctrl runtime (#506)

* feat: init ctrl runtime manager and use klogr

Signed-off-by: Wei Zhang <[email protected]>

* feat: migrate podgroup to ctrl runtime

Signed-off-by: Wei Zhang <[email protected]>

* feat: drop legacy podGroup controller

Signed-off-by: Wei Zhang <[email protected]>

* test: add ctrl runtime podGroup controller test

Signed-off-by: Wei Zhang <[email protected]>

* feat: drop cosheduler dir and fix manager not running

Signed-off-by: Wei Zhang <[email protected]>

---------

Signed-off-by: Wei Zhang <[email protected]>

Author: zwpaper

cmd/controller/app/options.go

@@ -23,6 +23,8 @@ import (
 type ServerRunOptions struct {
 	KubeConfig           string
 	MasterUrl            string
+	MetricsAddr          string
+	ProbeAddr            string
 	InCluster            bool
 	ApiServerQPS         int
 	ApiServerBurst       int
@@ -40,6 +42,8 @@ func (s *ServerRunOptions) addAllFlags() {
 	pflag.BoolVar(&s.InCluster, "incluster", s.InCluster, "If controller run incluster.")
 	pflag.StringVar(&s.KubeConfig, "kubeConfig", s.KubeConfig, "Kube Config path if not run in cluster.")
 	pflag.StringVar(&s.MasterUrl, "masterUrl", s.MasterUrl, "Master Url if not run in cluster.")
+	pflag.StringVar(&s.MetricsAddr, "metricsAddr", ":8080", "Metrics server bind listen address.")
+	pflag.StringVar(&s.ProbeAddr, "probeAddr", ":8081", "Probe endpoint bind  address.")
 	pflag.IntVar(&s.ApiServerQPS, "qps", 5, "qps of query apiserver.")
 	pflag.IntVar(&s.ApiServerBurst, "burst", 10, "burst of query apiserver.")
 	pflag.IntVar(&s.Workers, "workers", 1, "workers of scheduler-plugin-controllers.")

cmd/controller/app/server.go

@@ -20,22 +20,42 @@ import (
 	"context"
 	"os"
 
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/apiserver/pkg/server"
 	"k8s.io/client-go/informers"
 	"k8s.io/client-go/kubernetes"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	restclient "k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/clientcmd"
 	"k8s.io/client-go/tools/leaderelection"
 	"k8s.io/client-go/tools/leaderelection/resourcelock"
 	"k8s.io/klog/v2"
+	"k8s.io/klog/v2/klogr"
 
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/healthz"
+
+	schedulingv1a1 "sigs.k8s.io/scheduler-plugins/apis/scheduling/v1alpha1"
 	"sigs.k8s.io/scheduler-plugins/pkg/controller"
+	"sigs.k8s.io/scheduler-plugins/pkg/controllers"
 	schedclientset "sigs.k8s.io/scheduler-plugins/pkg/generated/clientset/versioned"
 	schedformers "sigs.k8s.io/scheduler-plugins/pkg/generated/informers/externalversions"
 )
 
+var (
+	scheme   = runtime.NewScheme()
+	setupLog = ctrl.Log.WithName("setup")
+)
+
+func init() {
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+
+	utilruntime.Must(schedulingv1a1.AddToScheme(scheme))
+}
+
 func newConfig(kubeconfig, master string, inCluster bool) (*restclient.Config, error) {
 	var (
 		config *rest.Config
@@ -66,17 +86,53 @@ func Run(s *ServerRunOptions) error {
 	kubeClient := kubernetes.NewForConfigOrDie(config)
 
 	schedInformerFactory := schedformers.NewSharedInformerFactory(schedClient, 0)
-	pgInformer := schedInformerFactory.Scheduling().V1alpha1().PodGroups()
 	eqInformer := schedInformerFactory.Scheduling().V1alpha1().ElasticQuotas()
 
 	coreInformerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
 	podInformer := coreInformerFactory.Core().V1().Pods()
-	pgCtrl := controller.NewPodGroupController(kubeClient, pgInformer, podInformer, schedClient)
 	eqCtrl := controller.NewElasticQuotaController(kubeClient, eqInformer, podInformer, schedClient)
 
+	// Controller Runtime Controllers
+	ctrl.SetLogger(klogr.New())
+	mgr, err := ctrl.NewManager(ctrl.GetConfigOrDie(), ctrl.Options{
+		Scheme:                  scheme,
+		MetricsBindAddress:      s.MetricsAddr,
+		Port:                    9443,
+		HealthProbeBindAddress:  s.ProbeAddr,
+		LeaderElection:          s.EnableLeaderElection,
+		LeaderElectionID:        "sched-plugins-controllers",
+		LeaderElectionNamespace: "kube-system",
+	})
+	if err != nil {
+		setupLog.Error(err, "unable to start manager")
+		return err
+	}
+
+	if err = (&controllers.PodGroupReconciler{
+		Client: mgr.GetClient(),
+		Scheme: mgr.GetScheme(),
+	}).SetupWithManager(mgr); err != nil {
+		setupLog.Error(err, "unable to create controller", "controller", "PodGroup")
+		return err
+	}
+
+	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
+		setupLog.Error(err, "unable to set up health check")
+		return err
+	}
+	if err := mgr.AddReadyzCheck("readyz", healthz.Ping); err != nil {
+		setupLog.Error(err, "unable to set up ready check")
+		return err
+	}
+
 	run := func(ctx context.Context) {
-		go pgCtrl.Run(s.Workers, ctx.Done())
 		go eqCtrl.Run(s.Workers, ctx.Done())
+		setupLog.Info("starting manager")
+		if err := mgr.Start(ctrl.SetupSignalHandler()); err != nil {
+			setupLog.Error(err, "unable to start manager")
+			panic(err)
+		}
+
 		select {}
 	}
 	schedInformerFactory.Start(stopCh)

go.mod

@@ -6,6 +6,7 @@ require (
 	github.com/diktyo-io/appgroup-api v0.0.9-alpha
 	github.com/diktyo-io/networktopology-api v0.0.8-alpha
 	github.com/dustin/go-humanize v1.0.0
+	github.com/go-logr/logr v1.2.3
 	github.com/google/go-cmp v0.5.8
 	github.com/k8stopologyawareschedwg/noderesourcetopology-api v0.0.13
 	github.com/k8stopologyawareschedwg/podfingerprint v0.1.1
@@ -48,7 +49,6 @@ require (
 	github.com/felixge/httpsnoop v1.0.1 // indirect
 	github.com/francoispqt/gojay v1.2.13 // indirect
 	github.com/fsnotify/fsnotify v1.5.1 // indirect
-	github.com/go-logr/logr v1.2.3 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/jsonreference v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.14 // indirect