refactor: enable 'revive.deep-exit' rule (#6030)

* chore: enable 'revive.deep-exit' rule

* fix deep-exit issue in pkg/controller/jobframework

* fix lint issue for yaml-processor

* fix lint issue for pkg/visibility

* fix lint issues

* fix compilation

* fix tests

* fix e2e

Author: alexandear

.golangci.yaml

@@ -64,6 +64,7 @@ linters:
       enable-all-rules: false
       rules:
         - name: context-as-argument
+        - name: deep-exit
         - name: empty-lines
         - name: increment-decrement
         - name: var-naming

cmd/kueue/main.go

@@ -21,6 +21,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"flag"
+	"fmt"
 	"net/http"
 	"os"
 	"path/filepath"
@@ -232,22 +233,43 @@ func main() {
 	}
 	debugger.NewDumper(cCache, queues).ListenForSignal(ctx)
 
-	serverVersionFetcher := setupServerVersionFetcher(mgr, kubeConfig)
+	serverVersionFetcher, err := setupServerVersionFetcher(mgr, kubeConfig)
+	if err != nil {
+		setupLog.Error(err, "Unable to setup server version fetcher")
+		os.Exit(1)
+	}
+
+	if err := setupProbeEndpoints(mgr, certsReady); err != nil {
+		setupLog.Error(err, "Unable to setup probe endpoints")
+		os.Exit(1)
+	}
 
-	setupProbeEndpoints(mgr, certsReady)
 	// Cert won't be ready until manager starts, so start a goroutine here which
 	// will block until the cert is ready before setting up the controllers.
 	// Controllers who register after manager starts will start directly.
-	go setupControllers(ctx, mgr, cCache, queues, certsReady, &cfg, serverVersionFetcher)
+	go func() {
+		if err := setupControllers(ctx, mgr, cCache, queues, certsReady, &cfg, serverVersionFetcher); err != nil {
+			setupLog.Error(err, "Unable to setup controllers")
+			os.Exit(1)
+		}
+	}()
 
 	go queues.CleanUpOnContext(ctx)
 	go cCache.CleanUpOnContext(ctx)
 
 	if features.Enabled(features.VisibilityOnDemand) {
-		go visibility.CreateAndStartVisibilityServer(ctx, queues)
+		go func() {
+			if err := visibility.CreateAndStartVisibilityServer(ctx, queues); err != nil {
+				setupLog.Error(err, "Unable to create and start visibility server")
+				os.Exit(1)
+			}
+		}()
 	}
 
-	setupScheduler(mgr, cCache, queues, &cfg)
+	if err := setupScheduler(mgr, cCache, queues, &cfg); err != nil {
+		setupLog.Error(err, "Could not setup scheduler")
+		os.Exit(1)
+	}
 
 	setupLog.Info("Starting manager")
 	if err := mgr.Start(ctx); err != nil {
@@ -267,22 +289,19 @@ func setupIndexes(ctx context.Context, mgr ctrl.Manager, cfg *configapi.Configur
 		if err := provisioning.ServerSupportsProvisioningRequest(mgr); err != nil {
 			setupLog.Error(err, "Skipping admission check controller setup: Provisioning Requests not supported (Possible cause: missing or unsupported cluster-autoscaler)")
 		} else if err := provisioning.SetupIndexer(ctx, mgr.GetFieldIndexer()); err != nil {
-			setupLog.Error(err, "Could not setup provisioning indexer")
-			os.Exit(1)
+			return fmt.Errorf("could not setup provisioning indexer: %w", err)
 		}
 	}
 
 	if features.Enabled(features.TopologyAwareScheduling) {
 		if err := tasindexer.SetupIndexes(ctx, mgr.GetFieldIndexer()); err != nil {
-			setupLog.Error(err, "Could not setup TAS indexer")
-			os.Exit(1)
+			return fmt.Errorf("could not setup TAX indexer: %w", err)
 		}
 	}
 
 	if features.Enabled(features.MultiKueue) {
 		if err := multikueue.SetupIndexer(ctx, mgr.GetFieldIndexer(), *cfg.Namespace); err != nil {
-			setupLog.Error(err, "Could not setup multikueue indexer")
-			os.Exit(1)
+			return fmt.Errorf("could not setup multikueue indexer: %w", err)
 		}
 	}
 
@@ -292,14 +311,13 @@ func setupIndexes(ctx context.Context, mgr ctrl.Manager, cfg *configapi.Configur
 	return jobframework.SetupIndexes(ctx, mgr.GetFieldIndexer(), opts...)
 }
 
-func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manager, certsReady chan struct{}, cfg *configapi.Configuration, serverVersionFetcher *kubeversion.ServerVersionFetcher) {
+func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manager, certsReady chan struct{}, cfg *configapi.Configuration, serverVersionFetcher *kubeversion.ServerVersionFetcher) error {
 	// The controllers won't work until the webhooks are operating, and the webhook won't work until the
 	// certs are all in place.
 	cert.WaitForCertsReady(setupLog, certsReady)
 
 	if failedCtrl, err := core.SetupControllers(mgr, queues, cCache, cfg); err != nil {
-		setupLog.Error(err, "Unable to create controller", "controller", failedCtrl)
-		os.Exit(1)
+		return fmt.Errorf("unable to create controller %s: %w", failedCtrl, err)
 	}
 
 	// setup provision admission check controller
@@ -309,44 +327,38 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache
 		} else {
 			ctrl, err := provisioning.NewController(mgr.GetClient(), mgr.GetEventRecorderFor("kueue-provisioning-request-controller"))
 			if err != nil {
-				setupLog.Error(err, "Could not create the provisioning controller")
-				os.Exit(1)
+				return fmt.Errorf("could not create the provisioning controller: %w", err)
 			}
 
 			if err := ctrl.SetupWithManager(mgr); err != nil {
-				setupLog.Error(err, "Could not setup provisioning controller")
-				os.Exit(1)
+				return fmt.Errorf("could not setup provisioning controller: %w", err)
 			}
 		}
 	}
 
 	if features.Enabled(features.MultiKueue) {
 		adapters, err := jobframework.GetMultiKueueAdapters(sets.New(cfg.Integrations.Frameworks...))
 		if err != nil {
-			setupLog.Error(err, "Could not get the enabled multikueue adapters")
-			os.Exit(1)
+			return fmt.Errorf("could not get the enabled multikueue adapters: %w", err)
 		}
 		if err := multikueue.SetupControllers(mgr, *cfg.Namespace,
 			multikueue.WithGCInterval(cfg.MultiKueue.GCInterval.Duration),
 			multikueue.WithOrigin(ptr.Deref(cfg.MultiKueue.Origin, configapi.DefaultMultiKueueOrigin)),
 			multikueue.WithWorkerLostTimeout(cfg.MultiKueue.WorkerLostTimeout.Duration),
 			multikueue.WithAdapters(adapters),
 		); err != nil {
-			setupLog.Error(err, "Could not setup MultiKueue controller")
-			os.Exit(1)
+			return fmt.Errorf("could not setup MultiKueue controller: %w", err)
 		}
 	}
 
 	if features.Enabled(features.TopologyAwareScheduling) {
 		if failedCtrl, err := tas.SetupControllers(mgr, queues, cCache, cfg); err != nil {
-			setupLog.Error(err, "Could not setup TAS controller", "controller", failedCtrl)
-			os.Exit(1)
+			return fmt.Errorf("could not setup TAS controller %s: %w", failedCtrl, err)
 		}
 	}
 
 	if failedWebhook, err := webhooks.Setup(mgr); err != nil {
-		setupLog.Error(err, "Unable to create webhook", "webhook", failedWebhook)
-		os.Exit(1)
+		return fmt.Errorf("unable to create webhook %s: %w", failedWebhook, err)
 	}
 
 	opts := []jobframework.Option{
@@ -366,24 +378,23 @@ func setupControllers(ctx context.Context, mgr ctrl.Manager, cCache *cache.Cache
 	}
 	nsSelector, err := metav1.LabelSelectorAsSelector(cfg.ManagedJobsNamespaceSelector)
 	if err != nil {
-		setupLog.Error(err, "Failed to parse managedJobsNamespaceSelector")
-		os.Exit(1)
+		return fmt.Errorf("failed to parse managedJobsNamespaceSelector: %w", err)
 	}
 	opts = append(opts, jobframework.WithManagedJobsNamespaceSelector(nsSelector))
 
 	if err := jobframework.SetupControllers(ctx, mgr, setupLog, opts...); err != nil {
-		setupLog.Error(err, "Unable to create controller or webhook", "kubernetesVersion", serverVersionFetcher.GetServerVersion())
-		os.Exit(1)
+		return fmt.Errorf("unable to create controller or webhook for kubernetesVersion %v: %w", serverVersionFetcher.GetServerVersion(), err)
 	}
+
+	return nil
 }
 
 // setupProbeEndpoints registers the health endpoints
-func setupProbeEndpoints(mgr ctrl.Manager, certsReady <-chan struct{}) {
+func setupProbeEndpoints(mgr ctrl.Manager, certsReady <-chan struct{}) error {
 	defer setupLog.Info("Probe endpoints are configured on healthz and readyz")
 
 	if err := mgr.AddHealthzCheck("healthz", healthz.Ping); err != nil {
-		setupLog.Error(err, "unable to set up health check")
-		os.Exit(1)
+		return fmt.Errorf("unable to set up health check: %w", err)
 	}
 
 	// Wait for the webhook server to be listening before advertising the
@@ -401,12 +412,13 @@ func setupProbeEndpoints(mgr ctrl.Manager, certsReady <-chan struct{}) {
 			return errors.New("certificates are not ready")
 		}
 	}); err != nil {
-		setupLog.Error(err, "unable to set up ready check")
-		os.Exit(1)
+		return fmt.Errorf("unable to set up ready check: %w", err)
 	}
+
+	return nil
 }
 
-func setupScheduler(mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manager, cfg *configapi.Configuration) {
+func setupScheduler(mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manager, cfg *configapi.Configuration) error {
 	sched := scheduler.New(
 		queues,
 		cCache,
@@ -416,31 +428,28 @@ func setupScheduler(mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manager
 		scheduler.WithFairSharing(cfg.FairSharing),
 	)
 	if err := mgr.Add(sched); err != nil {
-		setupLog.Error(err, "Unable to add scheduler to manager")
-		os.Exit(1)
+		return fmt.Errorf("unable to add scheduler to manager: %w", err)
 	}
+	return nil
 }
 
-func setupServerVersionFetcher(mgr ctrl.Manager, kubeConfig *rest.Config) *kubeversion.ServerVersionFetcher {
+func setupServerVersionFetcher(mgr ctrl.Manager, kubeConfig *rest.Config) (*kubeversion.ServerVersionFetcher, error) {
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(kubeConfig)
 	if err != nil {
-		setupLog.Error(err, "Unable to create the discovery client")
-		os.Exit(1)
+		return nil, fmt.Errorf("unable to create the discovery client: %w", err)
 	}
 
 	serverVersionFetcher := kubeversion.NewServerVersionFetcher(discoveryClient)
 
 	if err := mgr.Add(serverVersionFetcher); err != nil {
-		setupLog.Error(err, "Unable to add server version fetcher to manager")
-		os.Exit(1)
+		return nil, fmt.Errorf("unable to add server version fetcher to manager: %w", err)
 	}
 
 	if err := serverVersionFetcher.FetchServerVersion(); err != nil {
-		setupLog.Error(err, "failed to fetch kubernetes server version")
-		os.Exit(1)
+		return nil, fmt.Errorf("failed to fetch kubernetes server version: %w", err)
 	}
 
-	return serverVersionFetcher
+	return serverVersionFetcher, nil
 }
 
 func blockForPodsReady(cfg *configapi.Configuration) bool {

hack/internal/tools/yaml-processor/main.go

@@ -21,7 +21,6 @@ import (
 	"flag"
 	"fmt"
 	"log"
-	"os"
 	"syscall"
 
 	"go.uber.org/zap"
@@ -36,7 +35,10 @@ type options struct {
 }
 
 func main() {
-	opts := parseOptions()
+	opts, err := parseOptions()
+	if err != nil {
+		log.Fatalf("Failed to parse options: %v", err)
+	}
 
 	logger, err := newLogger(opts.LogLevel)
 	if err != nil {
@@ -65,7 +67,7 @@ func main() {
 	fileProcessor.ProcessPlan(*processingPlan)
 }
 
-func parseOptions() *options {
+func parseOptions() (*options, error) {
 	opts := &options{}
 
 	flag.StringVar(&opts.LogLevel, "zap-log-level", "info", "Minimum enabled logging level")
@@ -77,12 +79,12 @@ func parseOptions() *options {
 	flag.Parse()
 	if flag.NArg() != 1 {
 		flag.Usage()
-		os.Exit(1)
+		return nil, errors.New("exactly one processing plan file argument is required")
 	}
 
 	opts.ProcessingPlan = flag.Arg(0)
 
-	return opts
+	return opts, nil
 }
 
 func newLogger(logLevel string) (*zap.Logger, error) {

pkg/controller/jobframework/setup.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"errors"
 	"fmt"
-	"os"
 	"time"
 
 	"github.com/go-logr/logr"
@@ -73,8 +72,7 @@ func (m *integrationManager) setupControllers(ctx context.Context, mgr ctrl.Mana
 		if options.EnabledFrameworks.Has(name) {
 			if cb.CanSupportIntegration != nil {
 				if canSupport, err := cb.CanSupportIntegration(opts...); !canSupport || err != nil {
-					log.Error(err, "Failed to configure reconcilers")
-					os.Exit(1)
+					return fmt.Errorf("failed to configure reconcilers: %w", err)
 				}
 			}
 			gvk, err := apiutil.GVKForObject(cb.JobType, mgr.GetScheme())

pkg/visibility/server.go

@@ -20,7 +20,6 @@ import (
 	"context"
 	"fmt"
 	"net"
-	"os"
 	"strings"
 
 	validatingadmissionpolicy "k8s.io/apiserver/pkg/admission/plugin/policy/validating"
@@ -32,7 +31,6 @@ import (
 	genericoptions "k8s.io/apiserver/pkg/server/options"
 	"k8s.io/component-base/compatibility"
 	"k8s.io/component-base/version"
-	ctrl "sigs.k8s.io/controller-runtime"
 
 	generatedopenapi "sigs.k8s.io/kueue/apis/visibility/openapi"
 	visibilityv1beta1 "sigs.k8s.io/kueue/apis/visibility/v1beta1"
@@ -43,7 +41,6 @@ import (
 )
 
 var (
-	setupLog = ctrl.Log.WithName("visibility-server")
 	// Admission plugins that are enabled by default in the kubeapi server
 	// but are not required for the visibility server.
 	disabledPlugins = []string{
@@ -60,28 +57,26 @@ var (
 // +kubebuilder:rbac:groups=flowcontrol.apiserver.k8s.io,resources=flowschemas/status,verbs=patch
 
 // CreateAndStartVisibilityServer creates visibility server injecting KueueManager and starts it
-func CreateAndStartVisibilityServer(ctx context.Context, kueueMgr *queue.Manager) {
+func CreateAndStartVisibilityServer(ctx context.Context, kueueMgr *queue.Manager) error {
 	config := newVisibilityServerConfig()
 	if err := applyVisibilityServerOptions(config); err != nil {
-		setupLog.Error(err, "Unable to apply VisibilityServerOptions")
-		os.Exit(1)
+		return fmt.Errorf("unable to apply VisibilityServerOptions: %w", err)
 	}
 
 	visibilityServer, err := config.Complete().New("visibility-server", genericapiserver.NewEmptyDelegate())
 	if err != nil {
-		setupLog.Error(err, "Unable to create visibility server")
-		os.Exit(1)
+		return fmt.Errorf("unable to create visibility server: %w", err)
 	}
 
 	if err := api.Install(visibilityServer, kueueMgr); err != nil {
-		setupLog.Error(err, "Unable to install visibility.kueue.x-k8s.io API")
-		os.Exit(1)
+		return fmt.Errorf("unable to install visibility.kueue.x-k8s.io API: %w", err)
 	}
 
 	if err := visibilityServer.PrepareRun().RunWithContext(ctx); err != nil {
-		setupLog.Error(err, "Error running visibility server")
-		os.Exit(1)
+		return fmt.Errorf("running visibility server: %w", err)
 	}
+
+	return nil
 }
 
 func applyVisibilityServerOptions(config *genericapiserver.RecommendedConfig) error {

test/e2e/certmanager/suite_test.go

@@ -53,7 +53,9 @@ func TestAPIs(t *testing.T) {
 var _ = ginkgo.BeforeSuite(func() {
 	util.SetupLogger()
 
-	k8sClient, cfg = util.CreateClientUsingCluster("")
+	var err error
+	k8sClient, cfg, err = util.CreateClientUsingCluster("")
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	restClient = util.CreateRestClient(cfg)
 	ctx = ginkgo.GinkgoT().Context()
 

test/e2e/customconfigs/suite_test.go

@@ -56,7 +56,9 @@ func TestAPIs(t *testing.T) {
 var _ = ginkgo.BeforeSuite(func() {
 	util.SetupLogger()
 
-	k8sClient, cfg = util.CreateClientUsingCluster("")
+	var err error
+	k8sClient, cfg, err = util.CreateClientUsingCluster("")
+	gomega.Expect(err).NotTo(gomega.HaveOccurred())
 	restClient = util.CreateRestClient(cfg)
 	ctx = ginkgo.GinkgoT().Context()