Merge pull request #1285 from Tal-or/scheduler_ha

CNF-17778: CNF-17779: schedulercontroller: High availability by default

Author: openshift-merge-bot[bot]

api/v1/numaresourcesscheduler_defaults.go

@@ -28,7 +28,6 @@ const (
 	defaultSchedulerInformer    = SchedulerInformerDedicated
 	defaultCacheResyncDetection = CacheResyncDetectionRelaxed
 	defaultScoringStrategy      = LeastAllocated
-	defaultReplicas             = int32(1)
 )
 
 func SetDefaults_NUMAResourcesSchedulerSpec(spec *NUMAResourcesSchedulerSpec) {
@@ -54,8 +53,6 @@ func SetDefaults_NUMAResourcesSchedulerSpec(spec *NUMAResourcesSchedulerSpec) {
 			Type: defaultScoringStrategy,
 		}
 	}
-	if spec.Replicas == nil {
-		replicas := defaultReplicas
-		spec.Replicas = &replicas
-	}
+	// do not set replicas default value
+	// it will be computed later
 }

api/v1/numaresourcesscheduler_defaults_test.go

@@ -62,7 +62,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -83,7 +82,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -107,7 +105,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -132,7 +129,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -157,7 +153,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -180,7 +175,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				SchedulerInformer:    &schedInformer,
 				CacheResyncDetection: &cacheResyncDetection,
 				ScoringStrategy:      &scoringStrategyCustom,
-				Replicas:             ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -274,7 +268,6 @@ func TestSetDefaults_NUMAResourcesSchedulerSpec(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{

api/v1/numaresourcesscheduler_normalize_test.go

@@ -62,7 +62,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -83,7 +82,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -107,7 +105,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -132,7 +129,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -157,7 +153,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -180,7 +175,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				SchedulerInformer:    &schedInformer,
 				CacheResyncDetection: &cacheResyncDetection,
 				ScoringStrategy:      &scoringStrategyCustom,
-				Replicas:             ptr.To[int32](defaultReplicas),
 			},
 		},
 		{
@@ -274,7 +268,6 @@ func TestNUMAResourcesSchedulerSpecNormalize(t *testing.T) {
 				ScoringStrategy: &ScoringStrategyParams{
 					Type: scoringStrategyType,
 				},
-				Replicas: ptr.To[int32](defaultReplicas),
 			},
 		},
 		{

cmd/main.go

@@ -61,7 +61,6 @@ import (
 	"github.com/openshift-kni/numaresources-operator/pkg/hash"
 	"github.com/openshift-kni/numaresources-operator/pkg/images"
 	rtemetricsmanifests "github.com/openshift-kni/numaresources-operator/pkg/metrics/manifests/monitor"
-	"github.com/openshift-kni/numaresources-operator/pkg/numaresourcesscheduler/controlplane"
 	schedmanifests "github.com/openshift-kni/numaresources-operator/pkg/numaresourcesscheduler/manifests/sched"
 	rtestate "github.com/openshift-kni/numaresources-operator/pkg/objectstate/rte"
 	rteupdate "github.com/openshift-kni/numaresources-operator/pkg/objectupdate/rte"
@@ -159,7 +158,7 @@ func (pa *Params) FromFlags() {
 	flag.BoolVar(&pa.enableMCPCondsForward, "enable-mcp-conds-fwd", pa.enableMCPCondsForward, "enable MCP Status Condition forwarding")
 	flag.StringVar(&pa.image.Exporter, "image-exporter", pa.image.Exporter, "use this image as default for the RTE")
 	flag.StringVar(&pa.image.Scheduler, "image-scheduler", pa.image.Scheduler, "use this image as default for the scheduler")
-	flag.BoolVar(&pa.enableReplicasDetect, "detect-replicas", pa.enableReplicasDetect, "autodetect optimal replica count")
+	flag.BoolVar(&pa.enableReplicasDetect, "detect-replicas", pa.enableReplicasDetect, "autodetect optimal replica count.(DEPRECATED) autodetect enabled by default and should be configured from the NUMAResourcesScheduler CR")
 
 	flag.Parse()
 
@@ -312,11 +311,6 @@ func main() {
 	}
 
 	if params.enableScheduler {
-		info := controlplane.Defaults()
-		if params.enableReplicasDetect {
-			info = controlplane.Discover(ctx)
-		}
-
 		schedMf, err := schedmanifests.GetManifests(namespace)
 		if err != nil {
 			klog.ErrorS(err, "unable to load the Scheduler manifests")
@@ -329,7 +323,6 @@ func main() {
 			Scheme:             mgr.GetScheme(),
 			SchedulerManifests: schedMf,
 			Namespace:          namespace,
-			AutodetectReplicas: info.NodeCount,
 		}).SetupWithManager(mgr); err != nil {
 			klog.ErrorS(err, "unable to create controller", "controller", "NUMAResourcesScheduler")
 			os.Exit(1)

internal/controller/numaresourcesscheduler_controller.go

@@ -19,6 +19,7 @@ package controller
 import (
 	"context"
 	"fmt"
+	"sigs.k8s.io/controller-runtime/pkg/handler"
 	"time"
 
 	appsv1 "k8s.io/api/apps/v1"
@@ -27,8 +28,10 @@ import (
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	apierrors "k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/builder"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -60,7 +63,6 @@ type NUMAResourcesSchedulerReconciler struct {
 	Scheme             *runtime.Scheme
 	SchedulerManifests schedmanifests.Manifests
 	Namespace          string
-	AutodetectReplicas int
 }
 
 //+kubebuilder:rbac:groups=rbac.authorization.k8s.io,resources=clusterroles,verbs=*
@@ -126,6 +128,18 @@ func (r *NUMAResourcesSchedulerReconciler) SetupWithManager(mgr ctrl.Manager) er
 				!apiequality.Semantic.DeepEqual(e.ObjectNew.GetAnnotations(), e.ObjectOld.GetAnnotations())
 		},
 	}
+	nodesPredicate := predicate.Funcs{
+		// we only care about cases when nodes are getting created or deleted
+		CreateFunc: func(e event.TypedCreateEvent[client.Object]) bool {
+			return true
+		},
+		DeleteFunc: func(e event.TypedDeleteEvent[client.Object]) bool {
+			return true
+		},
+		UpdateFunc: func(e event.TypedUpdateEvent[client.Object]) bool {
+			return false
+		},
+	}
 
 	return ctrl.NewControllerManagedBy(mgr).
 		For(&nropv1.NUMAResourcesScheduler{}).
@@ -134,9 +148,25 @@ func (r *NUMAResourcesSchedulerReconciler) SetupWithManager(mgr ctrl.Manager) er
 		Owns(&corev1.ServiceAccount{}, builder.WithPredicates(p)).
 		Owns(&corev1.ConfigMap{}, builder.WithPredicates(p)).
 		Owns(&appsv1.Deployment{}, builder.WithPredicates(p)).
+		Watches(&corev1.Node{}, handler.EnqueueRequestsFromMapFunc(r.nodeToNUMAResourcesScheduler),
+			builder.WithPredicates(nodesPredicate)).
 		Complete(r)
 }
 
+func (r *NUMAResourcesSchedulerReconciler) nodeToNUMAResourcesScheduler(ctx context.Context, object client.Object) []reconcile.Request {
+	var requests []reconcile.Request
+	nross := &nropv1.NUMAResourcesSchedulerList{}
+	if err := r.List(ctx, nross); err != nil {
+		klog.ErrorS(err, "failed to List NUMAResourcesScheduler")
+	}
+	for _, instance := range nross.Items {
+		requests = append(requests, reconcile.Request{NamespacedName: client.ObjectKey{
+			Name: instance.Name,
+		}})
+	}
+	return requests
+}
+
 func (r *NUMAResourcesSchedulerReconciler) reconcileResource(ctx context.Context, instance *nropv1.NUMAResourcesScheduler) (reconcile.Result, string, error) {
 	schedStatus, err := r.syncNUMASchedulerResources(ctx, instance)
 	if err != nil {
@@ -171,13 +201,22 @@ func isDeploymentRunning(ctx context.Context, c client.Client, key nropv1.Namesp
 	return false, nil
 }
 
-func (r *NUMAResourcesSchedulerReconciler) computeSchedulerReplicas(schedSpec nropv1.NUMAResourcesSchedulerSpec) *int32 {
-	// the api validation/normalization layer must ensure this value is != nil
-	if *schedSpec.Replicas >= 0 { // 0 is legit value to disable the deployment
-		return schedSpec.Replicas
+func (r *NUMAResourcesSchedulerReconciler) computeSchedulerReplicas(ctx context.Context, schedSpec nropv1.NUMAResourcesSchedulerSpec) (*int32, error) {
+	// do not autodetect if explicitly set by the user
+	if schedSpec.Replicas != nil {
+		return schedSpec.Replicas, nil
+	}
+	nodeList := &corev1.NodeList{}
+	if err := r.Client.List(ctx, nodeList, &client.ListOptions{
+		LabelSelector: labels.SelectorFromSet(map[string]string{
+			"node-role.kubernetes.io/control-plane": "",
+		}),
+	}); err != nil {
+		return schedSpec.Replicas, err
 	}
-	v := int32(r.AutodetectReplicas)
-	return &v
+	replicas := ptr.To(int32(len(nodeList.Items)))
+	klog.InfoS("autodetect scheduler replicas", "replicas", *replicas)
+	return replicas, nil
 }
 
 func (r *NUMAResourcesSchedulerReconciler) syncNUMASchedulerResources(ctx context.Context, instance *nropv1.NUMAResourcesScheduler) (nropv1.NUMAResourcesSchedulerStatus, error) {
@@ -186,6 +225,11 @@ func (r *NUMAResourcesSchedulerReconciler) syncNUMASchedulerResources(ctx contex
 
 	schedSpec := instance.Spec.Normalize()
 	cacheResyncPeriod := unpackAPIResyncPeriod(schedSpec.CacheResyncPeriod)
+	replicas, err := r.computeSchedulerReplicas(ctx, schedSpec)
+	if err != nil {
+		return nropv1.NUMAResourcesSchedulerStatus{}, fmt.Errorf("failed to compute scheduler replicas: %w", err)
+	}
+	schedSpec.Replicas = replicas
 	params := configParamsFromSchedSpec(schedSpec, cacheResyncPeriod, r.Namespace)
 
 	schedName, ok := schedstate.SchedulerNameFromObject(r.SchedulerManifests.ConfigMap)
@@ -207,7 +251,7 @@ func (r *NUMAResourcesSchedulerReconciler) syncNUMASchedulerResources(ctx contex
 		},
 	}
 
-	r.SchedulerManifests.Deployment.Spec.Replicas = r.computeSchedulerReplicas(schedSpec)
+	r.SchedulerManifests.Deployment.Spec.Replicas = schedSpec.Replicas
 	klog.V(4).InfoS("using scheduler replicas", "replicas", *r.SchedulerManifests.Deployment.Spec.Replicas)
 	// TODO: if replicas doesn't make sense (autodetect disabled and user set impossible value) then we
 	// should set a degraded state

internal/controller/numaresourcesscheduler_controller_test.go

@@ -18,6 +18,7 @@ package controller
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/onsi/ginkgo/v2"
@@ -33,6 +34,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime"
 	"k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/klog/v2"
+	"k8s.io/utils/ptr"
 	"sigs.k8s.io/controller-runtime/pkg/client"
 	"sigs.k8s.io/controller-runtime/pkg/client/fake"
 	"sigs.k8s.io/controller-runtime/pkg/reconcile"
@@ -93,11 +95,14 @@ var _ = ginkgo.Describe("Test NUMAResourcesScheduler Reconcile", func() {
 	ginkgo.Context("with correct NRS CR", func() {
 		var nrs *nropv1.NUMAResourcesScheduler
 		var reconciler *NUMAResourcesSchedulerReconciler
+		numOfMasters := 3
 
 		ginkgo.BeforeEach(func() {
 			var err error
 			nrs = testobjs.NewNUMAResourcesScheduler("numaresourcesscheduler", "some/url:latest", testSchedulerName, 11*time.Second)
-			reconciler, err = NewFakeNUMAResourcesSchedulerReconciler(nrs)
+			initObjects := []runtime.Object{nrs}
+			initObjects = append(initObjects, fakeNodes(numOfMasters, 3)...)
+			reconciler, err = NewFakeNUMAResourcesSchedulerReconciler(initObjects...)
 			gomega.Expect(err).ToNot(gomega.HaveOccurred())
 		})
 
@@ -585,10 +590,20 @@ var _ = ginkgo.Describe("Test NUMAResourcesScheduler Reconcile", func() {
 		})
 
 		ginkgo.It("should set the leader election resource parameters by default", func() {
+			nrs := nrs.DeepCopy()
+			nrs.Spec.Replicas = ptr.To(int32(1))
+			gomega.Eventually(reconciler.Client.Update).WithArguments(context.TODO(), nrs).WithPolling(30 * time.Second).WithTimeout(5 * time.Minute).Should(gomega.Succeed())
+
+			_, err := reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: client.ObjectKeyFromObject(nrs)})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+			expectLeaderElectParams(reconciler.Client, false, testNamespace, nrosched.LeaderElectionResourceName)
+		})
+
+		ginkgo.It("should set the leader election resource parameters to true default", func() {
 			key := client.ObjectKeyFromObject(nrs)
 			_, err := reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
 			gomega.Expect(err).ToNot(gomega.HaveOccurred())
-			expectLeaderElectParams(reconciler.Client, false, testNamespace, nrosched.LeaderElectionResourceName)
+			expectLeaderElectParams(reconciler.Client, true, testNamespace, nrosched.LeaderElectionResourceName)
 		})
 
 		ginkgo.DescribeTable("should set the leader election resource parameters depending on replica count", func(replicas int32, expectedEnabled bool) {
@@ -605,6 +620,16 @@ var _ = ginkgo.Describe("Test NUMAResourcesScheduler Reconcile", func() {
 			ginkgo.Entry("replicas=1", int32(1), false),
 			ginkgo.Entry("replicas=3", int32(3), true),
 		)
+
+		ginkgo.It("should detect replicas number by default when spec.Replicas is unset", func() {
+			key := client.ObjectKeyFromObject(nrs)
+			_, err := reconciler.Reconcile(context.TODO(), reconcile.Request{NamespacedName: key})
+			gomega.Expect(err).ToNot(gomega.HaveOccurred())
+
+			dp := &appsv1.Deployment{}
+			gomega.Expect(reconciler.Client.Get(context.TODO(), client.ObjectKey{Namespace: testNamespace, Name: "secondary-scheduler"}, dp)).To(gomega.Succeed())
+			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)
+		})
 	})
 })
 
@@ -695,3 +720,28 @@ func expectLeaderElectParams(cli client.Client, enabled bool, resourceNamespace,
 	gomega.Expect(cfg.LeaderElection.ResourceNamespace).To(gomega.Equal(resourceNamespace))
 	gomega.Expect(cfg.LeaderElection.ResourceName).To(gomega.Equal(resourceName))
 }
+
+func fakeNodes(numOfMasters, numOfWorkers int) []runtime.Object {
+	var nodes []runtime.Object
+	for i := range numOfMasters {
+		nodes = append(nodes, &corev1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("master-node-%d", i+1),
+				Labels: map[string]string{
+					"node-role.kubernetes.io/control-plane": "",
+				},
+			},
+		})
+	}
+	for i := range numOfWorkers {
+		nodes = append(nodes, &corev1.Node{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: fmt.Sprintf("worker-node-%d", i+1),
+				Labels: map[string]string{
+					"node-role.kubernetes.io/worker": "",
+				},
+			},
+		})
+	}
+	return nodes
+}