feat(multikueue): implement KEP-9270 stepSize for Incremental Dispatcher

apis/config/v1beta1/configuration_types.go

@@ -338,6 +338,23 @@ type MultiKueue struct {
 	// GroupVersionKind (GVK) for MultiKueue operations.
 	// +optional
 	ExternalFrameworks []MultiKueueExternalFramework `json:"externalFrameworks,omitempty"`
+
+	// IncrementalDispatcherConfig contains the configuration for the incremental dispatcher.
+	// This field is only valid when DispatcherName is set to the incremental dispatcher.
+	// Note: This field is going to be ignored when the MultiKueueIncrementalDispatcherConfig feature gate is disabled.
+	// +optional
+	IncrementalDispatcherConfig *IncrementalDispatcherConfig `json:"incrementalDispatcherConfig,omitempty"`
+}
+
+// IncrementalDispatcherConfig holds configuration for the MultiKueue Incremental Dispatcher.
+type IncrementalDispatcherConfig struct {
+	// StepSize defines the number of worker clusters the Incremental Dispatcher
+	// will query simultaneously.
+	// Minimum value is 1. If not set, defaults to 3.
+	// +optional
+	// +kubebuilder:default=3
+	// +kubebuilder:validation:Minimum=1
+	StepSize *int32 `json:"stepSize,omitempty"`
 }
 
 // MultiKueueExternalFramework defines a framework that is not built-in.

apis/config/v1beta1/defaults.go

@@ -125,6 +125,8 @@ func SetDefaults_Configuration(cfg *Configuration) {
 	cfg.MultiKueue.Origin = new(cmp.Or(ptr.Deref(cfg.MultiKueue.Origin, ""), DefaultMultiKueueOrigin))
 	cfg.MultiKueue.WorkerLostTimeout = cmp.Or(cfg.MultiKueue.WorkerLostTimeout, &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout})
 	cfg.MultiKueue.DispatcherName = cmp.Or(cfg.MultiKueue.DispatcherName, new(MultiKueueDispatcherModeAllAtOnce))
+	cfg.MultiKueue.IncrementalDispatcherConfig = cmp.Or(cfg.MultiKueue.IncrementalDispatcherConfig, &IncrementalDispatcherConfig{})
+	cfg.MultiKueue.IncrementalDispatcherConfig.StepSize = cmp.Or(cfg.MultiKueue.IncrementalDispatcherConfig.StepSize, ptr.To[int32](3))
 
 	if fs := cfg.FairSharing; fs != nil && fs.Enable && len(fs.PreemptionStrategies) == 0 {
 		fs.PreemptionStrategies = []PreemptionStrategy{LessThanOrEqualToFinalShare, LessThanInitialShare}

apis/config/v1beta1/defaults_test.go

@@ -96,6 +96,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 		Origin:            new(DefaultMultiKueueOrigin),
 		WorkerLostTimeout: &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout},
 		DispatcherName:    new(MultiKueueDispatcherModeAllAtOnce),
+		IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+			StepSize: ptr.To[int32](3),
+		},
 	}
 
 	podsReadyTimeout := metav1.Duration{Duration: defaultPodsReadyTimeout}
@@ -512,6 +515,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
 					DispatcherName:    new(MultiKueueDispatcherModeIncremental),
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},
@@ -542,6 +548,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new(DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
 					DispatcherName:    defaultMultiKueue.DispatcherName,
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},
@@ -570,6 +579,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: 15 * time.Minute},
 					DispatcherName:    defaultMultiKueue.DispatcherName,
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				WaitForPodsReady:             &WaitForPodsReady{},

apis/config/v1beta2/configuration_types.go

@@ -334,6 +334,23 @@ type MultiKueue struct {
 	// ClusterProfile defines configuration for using the ClusterProfile API.
 	// +optional
 	ClusterProfile *ClusterProfile `json:"clusterProfile,omitempty"`
+
+	// IncrementalDispatcherConfig contains the configuration for the incremental dispatcher.
+	// This field is only valid when DispatcherName is set to the incremental dispatcher.
+	// Note: This field is going to be ignored when the MultiKueueIncrementalDispatcherConfig feature gate is disabled.
+	// +optional
+	IncrementalDispatcherConfig *IncrementalDispatcherConfig `json:"incrementalDispatcherConfig,omitempty"`
+}
+
+// IncrementalDispatcherConfig holds configuration for the MultiKueue Incremental Dispatcher.
+type IncrementalDispatcherConfig struct {
+	// StepSize defines the number of worker clusters the Incremental Dispatcher
+	// will query simultaneously.
+	// Minimum value is 1. If not set, defaults to 3.
+	// +optional
+	// +kubebuilder:default=3
+	// +kubebuilder:validation:Minimum=1
+	StepSize *int32 `json:"stepSize,omitempty"`
 }
 
 // MultiKueueExternalFramework defines a framework that is not built-in.

apis/config/v1beta2/defaults.go

@@ -123,6 +123,8 @@ func SetDefaults_Configuration(cfg *Configuration) {
 	cfg.MultiKueue.Origin = new(cmp.Or(ptr.Deref(cfg.MultiKueue.Origin, ""), DefaultMultiKueueOrigin))
 	cfg.MultiKueue.WorkerLostTimeout = cmp.Or(cfg.MultiKueue.WorkerLostTimeout, &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout})
 	cfg.MultiKueue.DispatcherName = cmp.Or(cfg.MultiKueue.DispatcherName, new(MultiKueueDispatcherModeAllAtOnce))
+	cfg.MultiKueue.IncrementalDispatcherConfig = cmp.Or(cfg.MultiKueue.IncrementalDispatcherConfig, &IncrementalDispatcherConfig{})
+	cfg.MultiKueue.IncrementalDispatcherConfig.StepSize = cmp.Or(cfg.MultiKueue.IncrementalDispatcherConfig.StepSize, ptr.To[int32](3))
 
 	if afs := cfg.AdmissionFairSharing; afs != nil {
 		afs.UsageSamplingInterval.Duration = cmp.Or(afs.UsageSamplingInterval.Duration, 5*time.Minute)

apis/config/v1beta2/defaults_test.go

@@ -96,6 +96,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 		Origin:            new(DefaultMultiKueueOrigin),
 		WorkerLostTimeout: &metav1.Duration{Duration: DefaultMultiKueueWorkerLostTimeout},
 		DispatcherName:    new(MultiKueueDispatcherModeAllAtOnce),
+		IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+			StepSize: ptr.To[int32](3),
+		},
 	}
 
 	defaultVisibilityServer := &VisibilityServerConfiguration{
@@ -554,6 +557,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
 					DispatcherName:    new(MultiKueueDispatcherModeIncremental),
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				VisibilityServer:             defaultVisibilityServer,
@@ -584,6 +590,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new(DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: time.Minute},
 					DispatcherName:    defaultMultiKueue.DispatcherName,
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				VisibilityServer:             defaultVisibilityServer,
@@ -612,6 +621,9 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					Origin:            new("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: 15 * time.Minute},
 					DispatcherName:    defaultMultiKueue.DispatcherName,
+					IncrementalDispatcherConfig: &IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: defaultManagedJobsNamespaceSelector,
 				VisibilityServer:             defaultVisibilityServer,

pkg/config/config_test.go

@@ -416,6 +416,9 @@ objectRetentionPolicies:
 		Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 		WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
 		DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeAllAtOnce),
+		IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+			StepSize: ptr.To[int32](3),
+		},
 	}
 
 	defaultVisibility := &configapi.VisibilityServerConfiguration{
@@ -749,6 +752,9 @@ objectRetentionPolicies:
 					Origin:            new("multikueue-manager1"),
 					WorkerLostTimeout: &metav1.Duration{Duration: 10 * time.Minute},
 					DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeIncremental),
+					IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 					ClusterProfile: &configapi.ClusterProfile{
 						CredentialsProviders: []configapi.ClusterProfileCredentialsProvider{
 							{
@@ -948,6 +954,9 @@ webhook:
 					Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
 					DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeAllAtOnce),
+					IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: &metav1.LabelSelector{
 					MatchExpressions: []metav1.LabelSelectorRequirement{
@@ -998,6 +1007,9 @@ webhook:
 					Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
 					DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeAllAtOnce),
+					IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: &metav1.LabelSelector{
 					MatchExpressions: []metav1.LabelSelectorRequirement{
@@ -1048,6 +1060,9 @@ webhook:
 					Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
 					DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeAllAtOnce),
+					IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: &metav1.LabelSelector{
 					MatchExpressions: []metav1.LabelSelectorRequirement{
@@ -1094,6 +1109,9 @@ webhook:
 					Origin:            ptr.To(configapi.DefaultMultiKueueOrigin),
 					WorkerLostTimeout: &metav1.Duration{Duration: configapi.DefaultMultiKueueWorkerLostTimeout},
 					DispatcherName:    ptr.To(configapi.MultiKueueDispatcherModeAllAtOnce),
+					IncrementalDispatcherConfig: &configapi.IncrementalDispatcherConfig{
+						StepSize: ptr.To[int32](3),
+					},
 				},
 				ManagedJobsNamespaceSelector: &metav1.LabelSelector{
 					MatchExpressions: []metav1.LabelSelectorRequirement{
@@ -1245,6 +1263,9 @@ func TestEncode(t *testing.T) {
 					"origin":            "multikueue",
 					"workerLostTimeout": "15m0s",
 					"dispatcherName":    configapi.MultiKueueDispatcherModeAllAtOnce,
+					"incrementalDispatcherConfig": map[string]any{
+						"stepSize": int64(3),
+					},
 				},
 				"visibilityServer": map[string]any{
 					"bindPort": int64(8082),

pkg/controller/workloaddispatcher/controllers.go

@@ -34,7 +34,7 @@ func SetupControllers(mgr ctrl.Manager, cfg *configapi.Configuration, roleTracke
 		return "", err
 	}
 
-	idRec := NewIncrementalDispatcherReconciler(mgr.GetClient(), helper, roleTracker)
+	idRec := NewIncrementalDispatcherReconciler(mgr.GetClient(), helper, roleTracker, cfg)
 	err = idRec.SetupWithManager(mgr, cfg)
 	if err != nil {
 		return "multikueue-incremental-dispatcher", err