[multikueue] add support for custom multikueue controller

apis/kueue/v1alpha1/multikueue_types.go

@@ -59,6 +59,14 @@ type KubeConfig struct {
 }
 
 type MultiKueueClusterSpec struct {
+	// controllerName is name of the controller which will actually perform
+	// the checks. This is the name with which controller identifies with,
+	// not necessarily a K8S Pod or Deployment name. Cannot be empty.
+	// +kubebuilder:validation:Required
+	// +kubebuilder:default="kueue.x-k8s.io/multikueue"
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf", message="field is immutable"
+	ControllerName string `json:"controllerName"`
+
 	// Information how to connect to the cluster.
 	KubeConfig KubeConfig `json:"kubeConfig"`
 }

charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml

@@ -53,6 +53,16 @@ spec:
             type: object
           spec:
             properties:
+              controllerName:
+                default: kueue.x-k8s.io/multikueue
+                description: |-
+                  controllerName is name of the controller which will actually perform
+                  the checks. This is the name with which controller identifies with,
+                  not necessarily a K8S Pod or Deployment name. Cannot be empty.
+                type: string
+                x-kubernetes-validations:
+                - message: field is immutable
+                  rule: self == oldSelf
               kubeConfig:
                 description: Information how to connect to the cluster.
                 properties:
@@ -76,6 +86,7 @@ spec:
                 - locationType
                 type: object
             required:
+            - controllerName
             - kubeConfig
             type: object
           status:

cmd/kueue/main.go

@@ -256,10 +256,18 @@ func setupControllers(mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manag
 	}
 
 	if features.Enabled(features.MultiKueue) {
+		adapters, err := jobframework.GetMultiKueueAdapters()
+		if err != nil {
+			setupLog.Error(err, "Could not get the multikueue adapters")
+			os.Exit(1)
+		}
+
 		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.WithControllerName(kueuealpha.MultiKueueControllerName),
+			multikueue.WithAdapters(adapters),
 		); err != nil {
 			setupLog.Error(err, "Could not setup MultiKueue controller")
 			os.Exit(1)

config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml

@@ -38,6 +38,16 @@ spec:
             type: object
           spec:
             properties:
+              controllerName:
+                default: kueue.x-k8s.io/multikueue
+                description: |-
+                  controllerName is name of the controller which will actually perform
+                  the checks. This is the name with which controller identifies with,
+                  not necessarily a K8S Pod or Deployment name. Cannot be empty.
+                type: string
+                x-kubernetes-validations:
+                - message: field is immutable
+                  rule: self == oldSelf
               kubeConfig:
                 description: Information how to connect to the cluster.
                 properties:
@@ -61,6 +71,7 @@ spec:
                 - locationType
                 type: object
             required:
+            - controllerName
             - kubeConfig
             type: object
           status:

pkg/controller/admissionchecks/multikueue/admissioncheck.go

@@ -52,19 +52,20 @@ func newMultiKueueStoreHelper(c client.Client) (*multiKueueStoreHelper, error) {
 }
 
 // ACReconciler implements the reconciler for all the admission checks controlled by multikueue.
-// Its main task being to maintain the active state of the admission checks based on the heath
+// Its main task being to maintain the active state of the admission checks based on the health
 // of its referenced MultiKueueClusters.
 type ACReconciler struct {
-	client client.Client
-	helper *multiKueueStoreHelper
+	controllerName string
+	client         client.Client
+	helper         *multiKueueStoreHelper
 }
 
 var _ reconcile.Reconciler = (*ACReconciler)(nil)
 
 func (a *ACReconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) {
 	log := ctrl.LoggerFrom(ctx)
 	ac := &kueue.AdmissionCheck{}
-	if err := a.client.Get(ctx, req.NamespacedName, ac); err != nil || ac.Spec.ControllerName != kueuealpha.MultiKueueControllerName {
+	if err := a.client.Get(ctx, req.NamespacedName, ac); err != nil || ac.Spec.ControllerName != a.controllerName {
 		return reconcile.Result{}, client.IgnoreNotFound(err)
 	}
 
@@ -96,7 +97,8 @@ func (a *ACReconciler) Reconcile(ctx context.Context, req reconcile.Request) (re
 
 			if err != nil {
 				missingClusters = append(missingClusters, clusterName)
-			} else if !apimeta.IsStatusConditionTrue(cluster.Status.Conditions, kueuealpha.MultiKueueClusterActive) {
+			} else if !apimeta.IsStatusConditionTrue(cluster.Status.Conditions, kueuealpha.MultiKueueClusterActive) ||
+				ac.Spec.ControllerName != cluster.Spec.ControllerName {
 				inactiveClusters = append(inactiveClusters, clusterName)
 			}
 		}
@@ -166,10 +168,11 @@ func (a *ACReconciler) Reconcile(ctx context.Context, req reconcile.Request) (re
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=admissionchecks,verbs=get;list;watch
 // +kubebuilder:rbac:groups=kueue.x-k8s.io,resources=multikueueconfigs,verbs=get;list;watch
 
-func newACReconciler(c client.Client, helper *multiKueueStoreHelper) *ACReconciler {
+func newACReconciler(c client.Client, helper *multiKueueStoreHelper, so SetupOptions) *ACReconciler {
 	return &ACReconciler{
-		client: c,
-		helper: helper,
+		controllerName: so.controllerName,
+		client:         c,
+		helper:         helper,
 	}
 }
 

pkg/controller/admissionchecks/multikueue/admissioncheck_test.go

@@ -17,6 +17,7 @@ limitations under the License.
 package multikueue
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/google/go-cmp/cmp"
@@ -123,6 +124,7 @@ func TestReconcile(t *testing.T) {
 			},
 			clusters: []kueuealpha.MultiKueueCluster{
 				*utiltesting.MakeMultiKueueCluster("worker1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionFalse, "ByTest", "by test", 1).
 					Obj(),
 			},
@@ -156,9 +158,11 @@ func TestReconcile(t *testing.T) {
 			},
 			clusters: []kueuealpha.MultiKueueCluster{
 				*utiltesting.MakeMultiKueueCluster("worker1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionFalse, "ByTest", "by test", 1).
 					Obj(),
 				*utiltesting.MakeMultiKueueCluster("worker2").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionFalse, "ByTest", "by test", 1).
 					Obj(),
 			},
@@ -192,9 +196,11 @@ func TestReconcile(t *testing.T) {
 			},
 			clusters: []kueuealpha.MultiKueueCluster{
 				*utiltesting.MakeMultiKueueCluster("worker1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionFalse, "ByTest", "by test", 1).
 					Obj(),
 				*utiltesting.MakeMultiKueueCluster("worker2").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionTrue, "ByTest", "by test", 1).
 					Obj(),
 			},
@@ -228,6 +234,7 @@ func TestReconcile(t *testing.T) {
 			},
 			clusters: []kueuealpha.MultiKueueCluster{
 				*utiltesting.MakeMultiKueueCluster("worker1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
 					Active(metav1.ConditionTrue, "ByTest", "by test", 1).
 					Obj(),
 			},
@@ -247,10 +254,49 @@ func TestReconcile(t *testing.T) {
 					Obj(),
 			},
 		},
+		"non default controller name": {
+			reconcileFor: "ac1",
+			checks: []kueue.AdmissionCheck{
+				*utiltesting.MakeAdmissionCheck("ac1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
+					Parameters(kueuealpha.GroupVersion.Group, "MultiKueueConfig", "config1").
+					Generation(1).
+					Obj(),
+			},
+			configs: []kueuealpha.MultiKueueConfig{
+				*utiltesting.MakeMultiKueueConfig("config1").Clusters("worker1", "worker2").Obj(),
+			},
+			clusters: []kueuealpha.MultiKueueCluster{
+				*utiltesting.MakeMultiKueueCluster("worker1").
+					ControllerName("mkc1").
+					Active(metav1.ConditionTrue, "ByTest", "by test", 1).
+					Obj(),
+				*utiltesting.MakeMultiKueueCluster("worker2").
+					ControllerName("mkc1").
+					Active(metav1.ConditionTrue, "ByTest", "by test", 1).
+					Obj(),
+			},
+			wantChecks: []kueue.AdmissionCheck{
+				*utiltesting.MakeAdmissionCheck("ac1").
+					ControllerName(kueuealpha.MultiKueueControllerName).
+					Parameters(kueuealpha.GroupVersion.Group, "MultiKueueConfig", "config1").
+					SingleInstanceInClusterQueue(true, SingleInstanceReason, SingleInstanceMessage, 1).
+					ApplyToAllFlavors(true, FlavorIndependentCheckReason, FlavorIndependentCheckMessage, 1).
+					Condition(metav1.Condition{
+						Type:               kueue.AdmissionCheckActive,
+						Status:             metav1.ConditionFalse,
+						Reason:             "NoUsableClusters",
+						Message:            "Inactive clusters: [worker1 worker2]",
+						ObservedGeneration: 1,
+					}).
+					Obj(),
+			},
+		},
 	}
 
 	for name, tc := range cases {
 		t.Run(name, func(t *testing.T) {
+			fmt.Println(name)
 			builder, ctx := getClientBuilder()
 
 			builder = builder.WithLists(
@@ -266,7 +312,8 @@ func TestReconcile(t *testing.T) {
 			c := builder.Build()
 
 			helper, _ := newMultiKueueStoreHelper(c)
-			reconciler := newACReconciler(c, helper)
+			setupOptions := NewSetupOptions()
+			reconciler := newACReconciler(c, helper, *setupOptions)
 
 			_, gotErr := reconciler.Reconcile(ctx, reconcile.Request{NamespacedName: types.NamespacedName{Name: tc.reconcileFor}})
 			if diff := cmp.Diff(tc.wantError, gotErr); diff != "" {

pkg/controller/admissionchecks/multikueue/controllers.go

@@ -21,6 +21,7 @@ import (
 
 	ctrl "sigs.k8s.io/controller-runtime"
 
+	kueuealpha "sigs.k8s.io/kueue/apis/kueue/v1alpha1"
 	"sigs.k8s.io/kueue/pkg/constants"
 	"sigs.k8s.io/kueue/pkg/controller/jobframework"
 )
@@ -32,10 +33,12 @@ const (
 )
 
 type SetupOptions struct {
+	controllerName    string
 	gcInterval        time.Duration
 	origin            string
 	workerLostTimeout time.Duration
 	eventsBatchPeriod time.Duration
+	adapters          map[string]jobframework.MultiKueueAdapter
 }
 
 type SetupOption func(o *SetupOptions)
@@ -72,13 +75,41 @@ func WithEventsBatchPeriod(d time.Duration) SetupOption {
 	}
 }
 
-func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) error {
-	options := &SetupOptions{
+// WithControllerName - sets the controller name for which the multikueue
+// admission check match.
+func WithControllerName(controllerName string) SetupOption {
+	return func(o *SetupOptions) {
+		o.controllerName = controllerName
+	}
+}
+
+// WithAdapters - sets all the MultiKueue adaptors.
+func WithAdapters(adapters map[string]jobframework.MultiKueueAdapter) SetupOption {
+	return func(o *SetupOptions) {
+		o.adapters = adapters
+	}
+}
+
+// WithAdapters - sets or updates the adadpter of the MultiKueue adaptors.
+func WithAdapter(adapter jobframework.MultiKueueAdapter) SetupOption {
+	return func(o *SetupOptions) {
+		o.adapters[adapter.GVK().String()] = adapter
+	}
+}
+
+func NewSetupOptions() *SetupOptions {
+	return &SetupOptions{
 		gcInterval:        defaultGCInterval,
 		origin:            defaultOrigin,
 		workerLostTimeout: defaultWorkerLostTimeout,
 		eventsBatchPeriod: constants.UpdatesBatchPeriod,
+		controllerName:    kueuealpha.MultiKueueControllerName,
+		adapters:          make(map[string]jobframework.MultiKueueAdapter),
 	}
+}
+
+func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) error {
+	options := NewSetupOptions()
 
 	for _, o := range opts {
 		o(options)
@@ -95,23 +126,18 @@ func SetupControllers(mgr ctrl.Manager, namespace string, opts ...SetupOption) e
 		return err
 	}
 
-	adapters, err := jobframework.GetMultiKueueAdapters()
-	if err != nil {
-		return err
-	}
-
-	cRec := newClustersReconciler(mgr.GetClient(), namespace, options.gcInterval, options.origin, fsWatcher, adapters)
+	cRec := newClustersReconciler(mgr.GetClient(), namespace, *options, fsWatcher)
 	err = cRec.setupWithManager(mgr)
 	if err != nil {
 		return err
 	}
 
-	acRec := newACReconciler(mgr.GetClient(), helper)
+	acRec := newACReconciler(mgr.GetClient(), helper, *options)
 	err = acRec.setupWithManager(mgr)
 	if err != nil {
 		return err
 	}
 
-	wlRec := newWlReconciler(mgr.GetClient(), helper, cRec, options.origin, options.workerLostTimeout, options.eventsBatchPeriod, adapters)
+	wlRec := newWlReconciler(mgr.GetClient(), helper, cRec, *options)
 	return wlRec.setupWithManager(mgr)
 }