fix(scheduler): fail fast when feature gate discovery fails (#2050)

Signed-off-by: Huy Nguyen VN <hcnguyen@nvidia.com>

Author: hcnguyen-ai

.changes/unreleased/fixed-20260810-172251.yaml

@@ -0,0 +1,6 @@
+kind: Fixed
+body: |-
+  Scheduler and binder now fail fast when Kubernetes API discovery fails at startup, instead of silently disabling dynamic resource allocation and node resource topology for the lifetime of the process.
+custom:
+  Issue: "2038"
+  Author: hcnguyen-ai

cmd/binder/app/app.go

@@ -105,7 +105,10 @@ func New(options *Options, config *rest.Config) (*App, error) {
 	kubeClient := draversionawareclient.NewDRAAwareClient(kubernetes.NewForConfigOrDie(config))
 	informerFactory := informers.NewSharedInformerFactory(kubeClient, 0)
 
-	featuregates.SetDRAFeatureGate(kubeClient.Discovery())
+	if err := featuregates.SetDRAFeatureGate(kubeClient.Discovery()); err != nil {
+		setupLog.Error(err, "unable to determine dynamic resource allocation availability")
+		return nil, err
+	}
 
 	rrs := resourcereservation.NewService(options.FakeGPUNodes, clientWithWatch, options.ResourceReservationPodImage,
 		time.Duration(options.ResourceReservationAllocationTimeout)*time.Second,

cmd/snapshot-tool/main.go

@@ -93,7 +93,10 @@ func main() {
 		DiscoveryClient:             kubeClient.Discovery(),
 	}
 
-	schedulerCache := cache.New(schedulerCacheParams)
+	schedulerCache, err := cache.New(schedulerCacheParams)
+	if err != nil {
+		log.InfraLogger.Fatalf("Failed to create scheduler cache: %v", err)
+	}
 	stopCh := make(chan struct{})
 	schedulerCache.Run(stopCh)
 	schedulerCache.WaitForCacheSync(stopCh)

pkg/common/feature_gates/feature_gates.go

@@ -4,14 +4,14 @@
 package featuregates
 
 import (
+	"fmt"
 	"strconv"
 	"strings"
 	"sync/atomic"
 
 	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/version"
 	discovery "k8s.io/client-go/discovery"
-	"sigs.k8s.io/controller-runtime/pkg/log"
 )
 
 const (
@@ -27,8 +27,13 @@ const (
 // off to reflect server-side DRA availability.
 var dynamicResourcesEnabled atomic.Bool
 
-func SetDRAFeatureGate(discoveryClient discovery.DiscoveryInterface) {
-	dynamicResourcesEnabled.Store(IsDynamicResourcesEnabled(discoveryClient))
+func SetDRAFeatureGate(discoveryClient discovery.DiscoveryInterface) error {
+	enabled, err := IsDynamicResourcesEnabled(discoveryClient)
+	if err != nil {
+		return err
+	}
+	dynamicResourcesEnabled.Store(enabled)
+	return nil
 }
 
 // DynamicResourcesEnabled reports whether DRA was determined to be usable
@@ -47,8 +52,13 @@ func SetDynamicResourcesEnabledForTest(enabled bool) {
 
 var nodeResourceTopologyEnabled atomic.Bool
 
-func SetNodeResourceTopologyFeatureGate(discoveryClient discovery.DiscoveryInterface) {
-	nodeResourceTopologyEnabled.Store(IsNodeResourceTopologyEnabled(discoveryClient))
+func SetNodeResourceTopologyFeatureGate(discoveryClient discovery.DiscoveryInterface) error {
+	enabled, err := IsNodeResourceTopologyEnabled(discoveryClient)
+	if err != nil {
+		return err
+	}
+	nodeResourceTopologyEnabled.Store(enabled)
+	return nil
 }
 
 func NodeResourceTopologyEnabled() bool {
@@ -61,43 +71,36 @@ func SetNodeResourceTopologyEnabledForTest(enabled bool) {
 
 // IsNodeResourceTopologyEnabled reports whether the cluster serves the
 // topology.node.k8s.io API group (the NodeResourceTopology CRD).
-func IsNodeResourceTopologyEnabled(discoveryClient discovery.DiscoveryInterface) bool {
-	logger := log.Log.WithName("feature-gates")
-
+func IsNodeResourceTopologyEnabled(discoveryClient discovery.DiscoveryInterface) (bool, error) {
 	serverGroups, err := discoveryClient.ServerGroups()
 	if err != nil {
-		logger.Error(err, "Failed to get server groups")
-		return false
+		return false, fmt.Errorf("failed to get server groups: %w", err)
 	}
 
 	for _, group := range serverGroups.Groups {
 		if group.Name == nodeResourceTopologyGroup {
-			return true
+			return true, nil
 		}
 	}
-	return false
+	return false, nil
 }
 
-func IsDynamicResourcesEnabled(discoveryClient discovery.DiscoveryInterface) bool {
-	logger := log.Log.WithName("feature-gates")
-
+func IsDynamicResourcesEnabled(discoveryClient discovery.DiscoveryInterface) (bool, error) {
 	// Get API server version
 	serverVersion, err := discoveryClient.ServerVersion()
 	if err != nil {
-		logger.Error(err, "Failed to get server version")
-		return false
+		return false, fmt.Errorf("failed to get server version: %w", err)
 	}
 
 	// Check if the API server version is compatible with DRA
 	if !isCompatibleDRAVersion(serverVersion) {
-		return false
+		return false, nil
 	}
 
 	// Get supported API versions
 	serverGroups, err := discoveryClient.ServerGroups()
 	if err != nil {
-		logger.Error(err, "Failed to get server groups")
-		return false
+		return false, fmt.Errorf("failed to get server groups: %w", err)
 	}
 
 	found := false
@@ -110,17 +113,17 @@ func IsDynamicResourcesEnabled(discoveryClient discovery.DiscoveryInterface) boo
 		}
 	}
 	if !found {
-		return false
+		return false, nil
 	}
 
 	// Check if the DRA API group is supported
 	for _, groupVersion := range resourceGroup.Versions {
 		if version.CompareKubeAwareVersionStrings(groupVersion.Version, minimalSupportedVersion) >= 0 {
-			return true
+			return true, nil
 		}
 	}
 
-	return false
+	return false, nil
 }
 
 func isCompatibleDRAVersion(serverVersion *version.Info) bool {

pkg/common/feature_gates/feature_gates_test.go

@@ -4,6 +4,7 @@
 package featuregates
 
 import (
+	"errors"
 	"testing"
 
 	. "github.com/onsi/ginkgo/v2"
@@ -13,10 +14,42 @@ import (
 	resourcev1beta1 "k8s.io/api/resource/v1beta1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	version "k8s.io/apimachinery/pkg/version"
+	"k8s.io/client-go/discovery"
 	fakediscovery "k8s.io/client-go/discovery/fake"
 	"k8s.io/client-go/kubernetes/fake"
 )
 
+// erroringDiscovery fails the requested discovery call, standing in for an API
+// server that is briefly unreachable while a component starts up.
+type erroringDiscovery struct {
+	discovery.DiscoveryInterface
+	versionErr error
+	groupsErr  error
+}
+
+func (e *erroringDiscovery) ServerVersion() (*version.Info, error) {
+	if e.versionErr != nil {
+		return nil, e.versionErr
+	}
+	return e.DiscoveryInterface.ServerVersion()
+}
+
+func (e *erroringDiscovery) ServerGroups() (*metav1.APIGroupList, error) {
+	if e.groupsErr != nil {
+		return nil, e.groupsErr
+	}
+	return e.DiscoveryInterface.ServerGroups()
+}
+
+func discoveryForVersion(major, minor string) discovery.DiscoveryInterface {
+	fakeClient := fake.NewClientset()
+	fakeClient.Discovery().(*fakediscovery.FakeDiscovery).FakedServerVersion = &version.Info{
+		Major: major,
+		Minor: minor,
+	}
+	return fakeClient.Discovery()
+}
+
 func TestCache(t *testing.T) {
 	RegisterFailHandler(Fail)
 	RunSpecs(t, "Test cache")
@@ -46,5 +79,75 @@ var _ = Describe("New", func() {
 			Entry("edge case version (1.31) with resource API should not enable DRA", "1", "31", []string{resourcev1alhpa3.SchemeGroupVersion.String()}, false),
 			Entry("higher compatible version (1.35) with resource API should enable DRA", "1", "34", []string{resourcev1.SchemeGroupVersion.String()}, true),
 		)
+
+		It("should return an error when the server version cannot be retrieved", func() {
+			discoveryClient := &erroringDiscovery{
+				DiscoveryInterface: discoveryForVersion("1", "34"),
+				versionErr:         errors.New("connection refused"),
+			}
+
+			_, err := IsDynamicResourcesEnabled(discoveryClient)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("should return an error when the server groups cannot be retrieved", func() {
+			discoveryClient := &erroringDiscovery{
+				DiscoveryInterface: discoveryForVersion("1", "34"),
+				groupsErr:          errors.New("connection refused"),
+			}
+
+			_, err := IsDynamicResourcesEnabled(discoveryClient)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("should leave the gate untouched when discovery fails", func() {
+			SetDynamicResourcesEnabledForTest(true)
+			DeferCleanup(SetDynamicResourcesEnabledForTest, false)
+
+			discoveryClient := &erroringDiscovery{
+				DiscoveryInterface: discoveryForVersion("1", "34"),
+				versionErr:         errors.New("connection refused"),
+			}
+
+			Expect(SetDRAFeatureGate(discoveryClient)).NotTo(Succeed())
+			Expect(DynamicResourcesEnabled()).To(BeTrue())
+		})
+	})
+
+	Context("NodeResourceTopology Feature Gate", func() {
+		It("should report availability when the API group is served", func() {
+			fakeClient := fake.NewClientset()
+			fakeClient.Resources = append(fakeClient.Resources,
+				&metav1.APIResourceList{GroupVersion: nodeResourceTopologyGroup + "/v1alpha2"})
+
+			Expect(IsNodeResourceTopologyEnabled(fakeClient.Discovery())).To(BeTrue())
+		})
+
+		It("should report unavailability when the API group is absent", func() {
+			Expect(IsNodeResourceTopologyEnabled(fake.NewClientset().Discovery())).To(BeFalse())
+		})
+
+		It("should return an error when the server groups cannot be retrieved", func() {
+			discoveryClient := &erroringDiscovery{
+				DiscoveryInterface: fake.NewClientset().Discovery(),
+				groupsErr:          errors.New("connection refused"),
+			}
+
+			_, err := IsNodeResourceTopologyEnabled(discoveryClient)
+			Expect(err).To(HaveOccurred())
+		})
+
+		It("should leave the gate untouched when discovery fails", func() {
+			SetNodeResourceTopologyEnabledForTest(true)
+			DeferCleanup(SetNodeResourceTopologyEnabledForTest, false)
+
+			discoveryClient := &erroringDiscovery{
+				DiscoveryInterface: fake.NewClientset().Discovery(),
+				groupsErr:          errors.New("connection refused"),
+			}
+
+			Expect(SetNodeResourceTopologyFeatureGate(discoveryClient)).NotTo(Succeed())
+			Expect(NodeResourceTopologyEnabled()).To(BeTrue())
+		})
 	})
 })

pkg/scheduler/cache/cache.go

@@ -113,7 +113,7 @@ func registerSchedulerPodInformer(informerFactory informers.SharedInformerFactor
 }
 
 // New returns a Cache implementation.
-func New(schedulerCacheParams *SchedulerCacheParams) Cache {
+func New(schedulerCacheParams *SchedulerCacheParams) (Cache, error) {
 	return newSchedulerCache(schedulerCacheParams)
 }
 
@@ -164,7 +164,7 @@ type SchedulerCache struct {
 	K8sClusterPodAffinityInfo
 }
 
-func newSchedulerCache(schedulerCacheParams *SchedulerCacheParams) *SchedulerCache {
+func newSchedulerCache(schedulerCacheParams *SchedulerCacheParams) (*SchedulerCache, error) {
 	sc := &SchedulerCache{
 		schedulingNodePoolParams:  schedulerCacheParams.NodePoolParams,
 		restrictNodeScheduling:    schedulerCacheParams.RestrictNodeScheduling,
@@ -196,13 +196,16 @@ func newSchedulerCache(schedulerCacheParams *SchedulerCacheParams) *SchedulerCac
 	sc.informerFactory = informers.NewSharedInformerFactory(sc.kubeClient, 0)
 	registerSchedulerPodInformer(sc.informerFactory)
 	if err := setSchedulerPodTransform(sc.informerFactory.Core().V1().Pods().Informer()); err != nil {
-		log.InfraLogger.Errorf("Failed to set scheduler pod transform: %v", err)
-		return nil
+		return nil, fmt.Errorf("failed to set scheduler pod transform: %w", err)
 	}
 	sc.kubeAiSchedulerInformerFactory = kubeaischedulerinfo.NewSharedInformerFactory(sc.kubeAiSchedulerClient, 0)
 
-	featuregates.SetDRAFeatureGate(schedulerCacheParams.DiscoveryClient)
-	featuregates.SetNodeResourceTopologyFeatureGate(schedulerCacheParams.DiscoveryClient)
+	if err := featuregates.SetDRAFeatureGate(schedulerCacheParams.DiscoveryClient); err != nil {
+		return nil, fmt.Errorf("failed to determine dynamic resource allocation availability: %w", err)
+	}
+	if err := featuregates.SetNodeResourceTopologyFeatureGate(schedulerCacheParams.DiscoveryClient); err != nil {
+		return nil, fmt.Errorf("failed to determine node resource topology availability: %w", err)
+	}
 	if featuregates.NodeResourceTopologyEnabled() && schedulerCacheParams.NRTClient != nil {
 		sc.nrtInformerFactory = nrtinformers.NewSharedInformerFactory(schedulerCacheParams.NRTClient, 0)
 	}
@@ -223,12 +226,11 @@ func newSchedulerCache(schedulerCacheParams *SchedulerCacheParams) *SchedulerCac
 		sc.restrictNodeScheduling, &sc.K8sClusterPodAffinityInfo, sc.scheduleCSIStorage, sc.fullHierarchyFairness, sc.StatusUpdater, sc.stuckInReleasingThreshold)
 
 	if err != nil {
-		log.InfraLogger.Errorf("Failed to create cluster info object: %v", err)
-		return nil
+		return nil, fmt.Errorf("failed to create cluster info object: %w", err)
 	}
 	sc.clusterInfo = clusterInfo
 
-	return sc
+	return sc, nil
 }
 
 func (sc *SchedulerCache) Snapshot() (*api.ClusterInfo, error) {

pkg/scheduler/cache/cache_test.go

@@ -60,12 +60,13 @@ var _ = Describe("Cache", func() {
 		Context("Pod informer filtering", func() {
 			It("should filter failed pods while keeping succeeded pods, without filtering by scheduler name", func() {
 				kubeClient := fake.NewSimpleClientset()
-				cache := New(&SchedulerCacheParams{
+				cache, err := New(&SchedulerCacheParams{
 					KubeClient:         kubeClient,
 					KAISchedulerClient: kubeaischedulerfake.NewSimpleClientset(),
 					NodePoolParams:     &conf.SchedulingNodePoolParams{},
 					DiscoveryClient:    kubeClient.Discovery(),
 				})
+				Expect(err).NotTo(HaveOccurred())
 
 				stopCh := make(chan struct{})
 				defer close(stopCh)
@@ -123,8 +124,9 @@ var _ = Describe("Cache", func() {
 						DiscoveryClient:    fakeClient.Discovery(),
 					}
 
-					cache := New(params)
+					cache, err := New(params)
 
+					Expect(err).NotTo(HaveOccurred())
 					Expect(cache).NotTo(BeNil())
 					Expect(featuregates.DynamicResourcesEnabled()).To(Equal(expectDRAAvailable))
 				},
@@ -515,13 +517,14 @@ func setupCacheWithObjects(snapshot bool, objects []runtime.Object, kaiScheduler
 	kubeClient := fake.NewSimpleClientset(objects...)
 	kubeAiSchedulerClient := kubeaischedulerfake.NewSimpleClientset(kaiSchedulerObjects...)
 
-	cache := New(&SchedulerCacheParams{
+	cache, err := New(&SchedulerCacheParams{
 		KubeClient:            kubeClient,
 		KAISchedulerClient:    kubeAiSchedulerClient,
 		NodePoolParams:        &conf.SchedulingNodePoolParams{},
 		FullHierarchyFairness: true,
 		DiscoveryClient:       kubeClient.Discovery(),
 	})
+	Expect(err).NotTo(HaveOccurred())
 
 	stopCh := make(chan struct{})
 	cache.Run(stopCh)