controller: sched: conditionally enable Shared informer by default

We are expecting c6c23c50f8dbdb863acb260f82f749262572abd3 to land to
openshift soon, once it lands it will be enabled by default and we want
the scheduler to adapt to this behavior by default iff the user didn't
explicitly set the informer mode in the CR.

Until now, the scheduler works in a dedicated informer mode which is meant to take into
account the pods in terminal state (such as pods that ran and completed)
in the PFP computation to ensure it matches the PFP computed by the RTE
and reported in NRT.

The intended behavior which the new kubelet behavior is about ignoring
such pods and accounting only for active pods, so if this behavior is
enabled in the RTE-NRT, while kept the default dedicated in the
scheduler there will be misalignment in the computed vs the expected PFP
from scheduler's POV vs NRT's POV and a scheduler stall will happen that
will never recover.

In this commit we adjust the informer default value to Shared (instead
of Dedicated) only if both below conditions are met:
1. the cluster version supports the fixed kubelet which is met if the
   cluster version is equal or greater to the known-to-be fixed OCP version.
2. the user didn't set the Spec.SchedulerInformer field in the NRS CR
This modification will enable the shared mode which in turn includes only
running pods (= active pods) in the PFP computation from POV allowing
ultimately PFP alignment with NRT.

Signed-off-by: Shereen Haj <[email protected]>

Author: shajmakh

api/v1/numaresourcesscheduler_types.go

@@ -38,7 +38,7 @@ const (
 type SchedulerInformerMode string
 
 const (
-	// SchedulerInformerDedicated makes the NodeResourceTopologyMatch plugin use the default framework informer.
+	// SchedulerInformerShared makes the NodeResourceTopologyMatch plugin use the default framework informer.
 	SchedulerInformerShared SchedulerInformerMode = "Shared"
 
 	// SchedulerInformerDedicated sets an additional separate informer just for the NodeResourceTopologyMatch plugin. Default.

cmd/main.go

@@ -338,6 +338,10 @@ func main() {
 			Scheme:             mgr.GetScheme(),
 			SchedulerManifests: schedMf,
 			Namespace:          namespace,
+			PlatformInfo: controller.PlatformInfo{
+				Platform: clusterPlatform,
+				Version:  clusterPlatformVersion,
+			},
 		}).SetupWithManager(mgr); err != nil {
 			klog.ErrorS(err, "unable to create controller", "controller", "NUMAResourcesScheduler")
 			os.Exit(1)

internal/controller/numaresourcesscheduler_controller.go

@@ -41,6 +41,7 @@ import (
 	"sigs.k8s.io/controller-runtime/pkg/predicate"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
 
+	"github.com/k8stopologyawareschedwg/deployer/pkg/deployer/platform"
 	k8swgmanifests "github.com/k8stopologyawareschedwg/deployer/pkg/manifests"
 	k8swgrbacupdate "github.com/k8stopologyawareschedwg/deployer/pkg/objectupdate/rbac"
 
@@ -58,12 +59,24 @@ import (
 	"github.com/openshift-kni/numaresources-operator/pkg/status"
 )
 
+const (
+	// ActivePodsResourcesSupportSince defines the OCP version which started to support the fixed kubelet
+	// in which the PodResourcesAPI lists the active pods by default
+	activePodsResourcesSupportSince = "4.20.999"
+)
+
+type PlatformInfo struct {
+	Platform platform.Platform
+	Version  platform.Version
+}
+
 // NUMAResourcesSchedulerReconciler reconciles a NUMAResourcesScheduler object
 type NUMAResourcesSchedulerReconciler struct {
 	client.Client
 	Scheme             *runtime.Scheme
 	SchedulerManifests schedmanifests.Manifests
 	Namespace          string
+	PlatformInfo       PlatformInfo
 }
 
 //+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles,verbs=*
@@ -225,6 +238,8 @@ func (r *NUMAResourcesSchedulerReconciler) syncNUMASchedulerResources(ctx contex
 	klog.V(4).Info("SchedulerSync start")
 	defer klog.V(4).Info("SchedulerSync stop")
 
+	platformNormalize(&instance.Spec, r.PlatformInfo)
+
 	schedSpec := instance.Spec.Normalize()
 	cacheResyncPeriod := unpackAPIResyncPeriod(schedSpec.CacheResyncPeriod)
 	replicas, err := r.computeSchedulerReplicas(ctx, schedSpec)
@@ -293,6 +308,27 @@ func (r *NUMAResourcesSchedulerReconciler) syncNUMASchedulerResources(ctx contex
 	return schedStatus, nil
 }
 
+func platformNormalize(spec *nropv1.NUMAResourcesSchedulerSpec, platInfo PlatformInfo) {
+	if platInfo.Platform != platform.OpenShift && platInfo.Platform != platform.HyperShift {
+		return
+	}
+
+	parsedVersion, _ := platform.ParseVersion(activePodsResourcesSupportSince)
+	ok, err := platInfo.Version.AtLeast(parsedVersion)
+	if err != nil {
+		klog.Infof("failed to compare version %v with %v, err %v", parsedVersion, platInfo.Version, err)
+		return
+	}
+
+	if !ok {
+		return
+	}
+
+	if spec.SchedulerInformer == nil {
+		spec.SchedulerInformer = ptr.To(nropv1.SchedulerInformerShared)
+		klog.V(4).InfoS("SchedulerInformer default is overridden", "Platform", platInfo.Platform, "PlatformVersion", platInfo.Version.String(), "SchedulerInformer", &spec.SchedulerInformer)
+	}
+}
 func (r *NUMAResourcesSchedulerReconciler) updateStatus(ctx context.Context, sched *nropv1.NUMAResourcesScheduler, condition string, reason string, message string) error {
 	sched.Status.Conditions, _ = status.UpdateConditions(sched.Status.Conditions, condition, reason, message)
 	if err := r.Client.Status().Update(ctx, sched); err != nil {

internal/controller/numaresourcesscheduler_controller_test.go

@@ -40,6 +40,7 @@ import (
 
 	configv1 "github.com/openshift/api/config/v1"
 
+	"github.com/k8stopologyawareschedwg/deployer/pkg/deployer/platform"
 	depmanifests "github.com/k8stopologyawareschedwg/deployer/pkg/manifests"
 	depobjupdate "github.com/k8stopologyawareschedwg/deployer/pkg/objectupdate"
 
@@ -632,6 +633,164 @@ var _ = ginkgo.Describe("Test NUMAResourcesScheduler Reconcile", func() {
 			gomega.Expect(*dp.Spec.Replicas).To(gomega.Equal(int32(numOfMasters)), "number of replicas is different than number of control-planes nodes; want=%d got=%d", numOfMasters, *dp.Spec.Replicas)
 		})
 	})
+
+	ginkgo.Context("with kubelet PodResourcesAPI listing active pods by default", func() {
+		var nrs *nropv1.NUMAResourcesScheduler
+		var reconciler *NUMAResourcesSchedulerReconciler
+		numOfMasters := 3
+
+		ginkgo.When("kubelet fix is enabled", func() {
+			fixedVersion, _ := platform.ParseVersion(activePodsResourcesSupportSince)
+
+			ginkgo.DescribeTable("should configure by default the informerMode to the expected when field is not set", func(reconcilerPlatInfo PlatformInfo, expectedInformer string) {
+				var err error
+				nrs = testobjs.NewNUMAResourcesScheduler("numaresourcesscheduler", "some/url:latest", testSchedulerName, 11*time.Second)
+				initObjects := []runtime.Object{nrs}
+				initObjects = append(initObjects, fakeNodes(numOfMasters, 3)...)
+				reconciler, err = NewFakeNUMAResourcesSchedulerReconciler(initObjects...)
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				reconciler.PlatformInfo = reconcilerPlatInfo
+
+				key := client.ObjectKeyFromObject(nrs)
+				_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				expectCacheParams(reconciler.Client, depmanifests.CacheResyncAutodetect, depmanifests.CacheResyncOnlyExclusiveResources, expectedInformer)
+			},
+				ginkgo.Entry("with fixed Openshift the default informer is Shared", PlatformInfo{
+					Platform: platform.OpenShift,
+					Version:  fixedVersion,
+				}, depmanifests.CacheInformerShared),
+				ginkgo.Entry("with fixed Hypershift the default informer is Shared", PlatformInfo{
+					Platform: platform.HyperShift,
+					Version:  fixedVersion,
+				}, depmanifests.CacheInformerShared),
+				ginkgo.Entry("with unknown platform the default informer is Dedicated (unchanged)", PlatformInfo{}, depmanifests.CacheInformerDedicated))
+
+			ginkgo.DescribeTable("should preserve informerMode value if set", func(reconcilerPlatInfo PlatformInfo) {
+				var err error
+				nrs = testobjs.NewNUMAResourcesScheduler("numaresourcesscheduler", "some/url:latest", testSchedulerName, 11*time.Second)
+				infMode := nropv1.SchedulerInformerDedicated
+				nrs.Spec.SchedulerInformer = &infMode
+				initObjects := []runtime.Object{nrs}
+				initObjects = append(initObjects, fakeNodes(numOfMasters, 3)...)
+				reconciler, err = NewFakeNUMAResourcesSchedulerReconciler(initObjects...)
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				reconciler.PlatformInfo = reconcilerPlatInfo
+
+				key := client.ObjectKeyFromObject(nrs)
+				_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+				expectCacheParams(reconciler.Client, depmanifests.CacheResyncAutodetect, depmanifests.CacheResyncOnlyExclusiveResources, string(infMode))
+			},
+				ginkgo.Entry("with Openshift", PlatformInfo{
+					Platform: platform.OpenShift,
+					Version:  fixedVersion,
+				}),
+				ginkgo.Entry("with Hypershift", PlatformInfo{
+					Platform: platform.HyperShift,
+					Version:  fixedVersion,
+				}),
+				ginkgo.Entry("with unknown platform", PlatformInfo{}))
+
+			ginkgo.DescribeTable("should allow to update the informerMode to be Dedicated after an overridden default", func(reconcilerPlatInfo PlatformInfo) {
+				var err error
+				nrs = testobjs.NewNUMAResourcesScheduler("numaresourcesscheduler", "some/url:latest", testSchedulerName, 11*time.Second)
+				initObjects := []runtime.Object{nrs}
+				initObjects = append(initObjects, fakeNodes(numOfMasters, 3)...)
+				reconciler, err = NewFakeNUMAResourcesSchedulerReconciler(initObjects...)
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				reconciler.PlatformInfo = reconcilerPlatInfo
+
+				key := client.ObjectKeyFromObject(nrs)
+				_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				// intentionally skip checking default value
+
+				// should query the object after reconcile because the defaults are overridden
+				gomega.Expect(reconciler.Client.Get(context.TODO(), key, nrs)).ToNot(gomega.HaveOccurred())
+
+				nrsUpdated := nrs.DeepCopy()
+				informerMode := nropv1.SchedulerInformerDedicated
+				nrsUpdated.Spec.SchedulerInformer = &informerMode
+				gomega.Eventually(func() bool {
+					if err := reconciler.Client.Update(context.TODO(), nrsUpdated); err != nil {
+						klog.Warningf("failed to update the scheduler object; err: %v", err)
+						return false
+					}
+					return true
+				}, 30*time.Second, 5*time.Second).Should(gomega.BeTrue())
+
+				_, err = reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
+				gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+				expectCacheParams(reconciler.Client, depmanifests.CacheResyncAutodetect, depmanifests.CacheResyncOnlyExclusiveResources, string(informerMode))
+			},
+				ginkgo.Entry("with Openshift", PlatformInfo{
+					Platform: platform.OpenShift,
+					Version:  fixedVersion,
+				}),
+				ginkgo.Entry("with Hypershift", PlatformInfo{
+					Platform: platform.HyperShift,
+					Version:  fixedVersion,
+				}))
+		})
+	})
+})
+
+var _ = ginkgo.Describe("Test scheduler spec PreNormalize", func() {
+	ginkgo.When("Spec.SchedulerInformer is not set by the user", func() {
+		ginkgo.It("should override default informer to Shared if kubelet is fixed - first supported zstream version", func() {
+			v, _ := platform.ParseVersion(activePodsResourcesSupportSince)
+			spec := nropv1.NUMAResourcesSchedulerSpec{}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(*spec.SchedulerInformer).To(gomega.Equal(nropv1.SchedulerInformerShared))
+		})
+
+		ginkgo.It("should override default informer to Shared if kubelet is fixed - version is greater than first supported (zstream)", func() {
+			v, _ := platform.ParseVersion("4.20.1000")
+			spec := nropv1.NUMAResourcesSchedulerSpec{}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(*spec.SchedulerInformer).To(gomega.Equal(nropv1.SchedulerInformerShared))
+		})
+
+		ginkgo.It("should override default informer to Shared if kubelet is fixed - version is greater than first supported (ystream)", func() {
+			v, _ := platform.ParseVersion("4.21.0")
+			spec := nropv1.NUMAResourcesSchedulerSpec{}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(*spec.SchedulerInformer).To(gomega.Equal(nropv1.SchedulerInformerShared))
+		})
+
+		ginkgo.It("should not override default informer if kubelet is not fixed - version is less than first supported (zstream)", func() {
+			// this is only for testing purposes as there is plan to backport the fix to older minor versions
+			// will need to remove this test if the fix is supported starting the first zstream of the release
+			v, _ := platform.ParseVersion("4.20.0")
+			spec := nropv1.NUMAResourcesSchedulerSpec{}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(spec.SchedulerInformer).To(gomega.BeNil())
+		})
+
+		ginkgo.It("should not override default informer if kubelet is not fixed - version is less than first supported (ystream)", func() {
+			v, _ := platform.ParseVersion("4.13.0")
+			spec := nropv1.NUMAResourcesSchedulerSpec{}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(spec.SchedulerInformer).To(gomega.BeNil())
+		})
+	})
+	ginkgo.When("Spec.SchedulerInformer is set by the user", func() {
+		ginkgo.It("should preserve informer value set by the user even if kubelet is fixed", func() {
+			v, _ := platform.ParseVersion(activePodsResourcesSupportSince)
+			spec := nropv1.NUMAResourcesSchedulerSpec{
+				SchedulerInformer: ptr.To(nropv1.SchedulerInformerDedicated),
+			}
+			platformNormalize(&spec, PlatformInfo{Platform: platform.OpenShift, Version: v})
+			gomega.Expect(*spec.SchedulerInformer).To(gomega.Equal(nropv1.SchedulerInformerDedicated))
+		})
+	})
 })
 
 func pop(m map[string]string, k string) string {