Wrap main in a controller-runtime Manager

Start refactoring cluster autoscaler's main to use Manager from
controller-runtime. For now, that's needed to use Client that's used by
CapacityQuota provider. In the future, main should be refactored,
so the manager handles leader election, metrics, healthcheck and pprof
servers and graceful shutdown. Additionally, that change allows us
to write controller-runtime style controllers.

Author: norbertcyran

cluster-autoscaler/main.go

@@ -26,7 +26,9 @@ import (
 	"time"
 
 	"github.com/spf13/pflag"
-
+	"k8s.io/apimachinery/pkg/runtime"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	cqv1alpha1 "k8s.io/autoscaler/cluster-autoscaler/apis/capacityquota/autoscaling.x-k8s.io/v1alpha1"
 	capacityclient "k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/client"
 	"k8s.io/autoscaler/cluster-autoscaler/capacitybuffer/common"
 	"k8s.io/autoscaler/cluster-autoscaler/config/flags"
@@ -41,7 +43,12 @@ import (
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/clustersnapshot/store"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/framework"
 	"k8s.io/autoscaler/cluster-autoscaler/simulator/scheduling"
+	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/kubernetes/pkg/features"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/cache"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+	metricsserver "sigs.k8s.io/controller-runtime/pkg/metrics/server"
 
 	"k8s.io/apimachinery/pkg/api/meta"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -85,6 +92,16 @@ import (
 	"k8s.io/klog/v2"
 )
 
+var (
+	scheme = runtime.NewScheme()
+)
+
+func init() {
+	utilruntime.Must(clientgoscheme.AddToScheme(scheme))
+	utilruntime.Must(cqv1alpha1.AddToScheme(scheme))
+	// TODO: add other CRDs
+}
+
 func registerSignalHandlers(autoscaler core.Autoscaler) {
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, os.Interrupt, os.Kill, syscall.SIGTERM, syscall.SIGQUIT)
@@ -100,7 +117,7 @@ func registerSignalHandlers(autoscaler core.Autoscaler) {
 	}()
 }
 
-func buildAutoscaler(ctx context.Context, debuggingSnapshotter debuggingsnapshot.DebuggingSnapshotter) (core.Autoscaler, *loop.LoopTrigger, error) {
+func buildAutoscaler(ctx context.Context, debuggingSnapshotter debuggingsnapshot.DebuggingSnapshotter, mgr manager.Manager) (core.Autoscaler, *loop.LoopTrigger, error) {
 	// Get AutoscalingOptions from flags.
 	autoscalingOptions := flags.AutoscalingOptions()
 
@@ -133,6 +150,8 @@ func buildAutoscaler(ctx context.Context, debuggingSnapshotter debuggingsnapshot
 		DeleteOptions:        deleteOptions,
 		DrainabilityRules:    drainabilityRules,
 		ScaleUpOrchestrator:  orchestrator.New(),
+		KubeClientNew:        mgr.GetClient(),
+		KubeCache:            mgr.GetCache(),
 	}
 
 	opts.Processors = ca_processors.DefaultProcessors(autoscalingOptions)
@@ -304,12 +323,29 @@ func run(healthCheck *metrics.HealthCheck, debuggingSnapshotter debuggingsnapsho
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	autoscaler, trigger, err := buildAutoscaler(ctx, debuggingSnapshotter)
+	restConfig := kube_util.GetKubeConfig(autoscalingOpts.KubeClientOpts)
+	mgr, err := ctrl.NewManager(restConfig, ctrl.Options{
+		Scheme: scheme,
+		Cache: cache.Options{
+			DefaultTransform: cache.TransformStripManagedFields(),
+		},
+		// TODO: migrate leader election, metrics, healthcheck, pprof servers to Manager
+		LeaderElection:         false,
+		Metrics:                metricsserver.Options{BindAddress: "0"},
+		HealthProbeBindAddress: "0",
+		PprofBindAddress:       "0",
+	})
+	if err != nil {
+		klog.Fatalf("Failed to create manager: %v", err)
+	}
+
+	autoscaler, trigger, err := buildAutoscaler(ctx, debuggingSnapshotter, mgr)
 	if err != nil {
 		klog.Fatalf("Failed to create autoscaler: %v", err)
 	}
 
 	// Register signal handlers for graceful shutdown.
+	// TODO: replace with ctrl.SetupSignalHandlers() and handle graceful shutdown with context
 	registerSignalHandlers(autoscaler)
 
 	// Start updating health check endpoint.
@@ -320,22 +356,42 @@ func run(healthCheck *metrics.HealthCheck, debuggingSnapshotter debuggingsnapsho
 		klog.Fatalf("Failed to autoscaler background components: %v", err)
 	}
 
-	// Autoscale ad infinitum.
-	if autoscalingOpts.FrequentLoopsEnabled {
-		// We need to have two timestamps because the scaleUp activity alternates between processing ProvisioningRequests,
-		// so we need to pass the older timestamp (previousRun) to trigger.Wait to run immediately if only one of the activities is productive.
-		lastRun := time.Now()
-		previousRun := time.Now()
-		for {
-			trigger.Wait(previousRun)
-			previousRun, lastRun = lastRun, time.Now()
-			loop.RunAutoscalerOnce(autoscaler, healthCheck, lastRun)
-		}
-	} else {
-		for {
-			time.Sleep(autoscalingOpts.ScanInterval)
-			loop.RunAutoscalerOnce(autoscaler, healthCheck, time.Now())
+	err = mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
+		// Autoscale ad infinitum.
+		if autoscalingOpts.FrequentLoopsEnabled {
+			// We need to have two timestamps because the scaleUp activity alternates between processing ProvisioningRequests,
+			// so we need to pass the older timestamp (previousRun) to trigger.Wait to run immediately if only one of the activities is productive.
+			lastRun := time.Now()
+			previousRun := time.Now()
+			for {
+				select {
+				case <-ctx.Done():
+					// TODO: handle graceful shutdown with context
+					return nil
+				default:
+					trigger.Wait(previousRun)
+					previousRun, lastRun = lastRun, time.Now()
+					loop.RunAutoscalerOnce(autoscaler, healthCheck, lastRun)
+				}
+			}
+		} else {
+			for {
+				select {
+				case <-ctx.Done():
+					// TODO: handle graceful shutdown with context
+					return nil
+				case <-time.After(autoscalingOpts.ScanInterval):
+					loop.RunAutoscalerOnce(autoscaler, healthCheck, time.Now())
+				}
+			}
 		}
+	}))
+	if err != nil {
+		klog.Fatalf("Failed to add runnable to manager: %v", err)
+	}
+
+	if err := mgr.Start(ctx); err != nil {
+		klog.Fatalf("Manager exited with error: %v", err)
 	}
 }
 
@@ -373,6 +429,7 @@ func main() {
 	if err := logsapi.ValidateAndApply(loggingConfig, featureGate); err != nil {
 		klog.Fatalf("Failed to validate and apply logging configuration: %v", err)
 	}
+	ctrl.SetLogger(klog.NewKlogr())
 
 	healthCheck := metrics.NewHealthCheck(autoscalingOpts.MaxInactivityTime, autoscalingOpts.MaxFailingTime)