[multikueue] Add garbage collector.

kubernetes-sigs · Jan 29, 2024 · f2cf51b · f2cf51b
1 parent 248255d
commit f2cf51b
Show file tree

Hide file tree

Showing 21 changed files with 564 additions and 27 deletions.
diff --git a/apis/config/v1beta1/configuration_types.go b/apis/config/v1beta1/configuration_types.go
@@ -67,6 +67,9 @@ type Configuration struct {
 	// QueueVisibility is configuration to expose the information about the top
 	// pending workloads.
 	QueueVisibility *QueueVisibility `json:"queueVisibility,omitempty"`
+
+	// MultiKueue controls the behaviour of the MultiKueue AdmissionCheck Controller.
+	MultiKueue *MultiKueue `json:"multiKueue,omitempty"`
 }
 
 type ControllerManager struct {
@@ -199,6 +202,13 @@ type WaitForPodsReady struct {
 	RequeuingTimestamp *RequeuingTimestamp `json:"requeuingTimestamp,omitempty"`
 }
 
+type MultiKueue struct {
+	// GCTimeout defines the timeout between two consecutive garbage collection runs.
+	// Defaults to 1min. If 0, the garbage collection is disabled.
+	// +optional
+	GCTimeout *metav1.Duration `json:"gcTimeout,omitempty"`
+}
+
 type RequeuingTimestamp string
 
 const (

diff --git a/apis/config/v1beta1/defaults.go b/apis/config/v1beta1/defaults.go
@@ -44,6 +44,7 @@ const (
 	DefaultQueueVisibilityUpdateIntervalSeconds int32   = 5
 	DefaultClusterQueuesMaxCount                int32   = 10
 	defaultJobFrameworkName                             = "batch/job"
+	DefaultMultiKueueGCTimeout                          = time.Minute
 )
 
 func getOperatorNamespace() string {
@@ -161,4 +162,11 @@ func SetDefaults_Configuration(cfg *Configuration) {
 	if cfg.Integrations.PodOptions.PodSelector == nil {
 		cfg.Integrations.PodOptions.PodSelector = &metav1.LabelSelector{}
 	}
+
+	if cfg.MultiKueue == nil {
+		cfg.MultiKueue = &MultiKueue{}
+	}
+	if cfg.MultiKueue.GCTimeout == nil {
+		cfg.MultiKueue.GCTimeout = &metav1.Duration{Duration: DefaultMultiKueueGCTimeout}
+	}
 }
diff --git a/apis/config/v1beta1/defaults_test.go b/apis/config/v1beta1/defaults_test.go
@@ -96,6 +96,8 @@ func TestSetDefaults_Configuration(t *testing.T) {
 		},
 	}
 
+	defaultMultiKueue := &MultiKueue{GCTimeout: &metav1.Duration{Duration: DefaultMultiKueueGCTimeout}}
+
 	podsReadyTimeoutTimeout := metav1.Duration{Duration: defaultPodsReadyTimeout}
 	podsReadyTimeoutOverwrite := metav1.Duration{Duration: time.Minute}
 
@@ -118,6 +120,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting ControllerManager": {
@@ -158,6 +161,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default ControllerManager": {
@@ -214,6 +218,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not set LeaderElectionID": {
@@ -254,6 +259,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting InternalCertManagement": {
@@ -271,6 +277,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default InternalCertManagement": {
@@ -289,6 +296,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default values in custom ClientConnection": {
@@ -314,6 +322,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				},
 				Integrations:    overwriteNamespaceIntegrations,
 				QueueVisibility: defaultQueueVisibility,
+				MultiKueue:      defaultMultiKueue,
 			},
 		},
 		"should default empty custom ClientConnection": {
@@ -333,6 +342,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting waitForPodsReady values": {
@@ -359,6 +369,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"set waitForPodsReady.blockAdmission to false when enable is false": {
@@ -385,6 +396,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"respecting provided waitForPodsReady values": {
@@ -413,6 +425,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"integrations": {
@@ -436,6 +449,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					PodOptions: defaultIntegrations.PodOptions,
 				},
 				QueueVisibility: defaultQueueVisibility,
+				MultiKueue:      defaultMultiKueue,
 			},
 		},
 		"queue visibility": {
@@ -464,6 +478,26 @@ func TestSetDefaults_Configuration(t *testing.T) {
 						MaxCount: 0,
 					},
 				},
+				MultiKueue: defaultMultiKueue,
+			},
+		},
+		"multiKueue": {
+			original: &Configuration{
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				MultiKueue: &MultiKueue{GCTimeout: &metav1.Duration{Duration: time.Second}},
+			},
+			want: &Configuration{
+				Namespace:         ptr.To(DefaultNamespace),
+				ControllerManager: defaultCtrlManagerConfigurationSpec,
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				ClientConnection: defaultClientConnection,
+				Integrations:     defaultIntegrations,
+				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       &MultiKueue{GCTimeout: &metav1.Duration{Duration: time.Second}},
 			},
 		},
 	}

diff --git a/apis/config/v1beta1/zz_generated.deepcopy.go b/apis/config/v1beta1/zz_generated.deepcopy.go
diff --git a/apis/kueue/v1alpha1/multikueue_types.go b/apis/kueue/v1alpha1/multikueue_types.go
@@ -50,6 +50,14 @@ type KubeConfig struct {
 type MultiKueueClusterSpec struct {
 	// Information how to connect to the cluster.
 	KubeConfig KubeConfig `json:"kubeConfig"`
+
+	// A label value used to track the creator of workloads in the worker cluster.
+	//
+	// +kubebuilder:validation:Optional
+	// +kubebuilder:validation:XValidation:rule="self == oldSelf",message="Value is immutable"
+	// +kubebuilder:validation:MaxLength=63
+	// +kubebuilder:default=multikueue
+	Origin string `json:"origin,omitempty"`
 }
 
 type MultiKueueClusterStatus struct {

diff --git a/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml b/charts/kueue/templates/crd/kueue.x-k8s.io_multikueueclusters.yaml
@@ -65,6 +65,15 @@ spec:
                 - location
                 - locationType
                 type: object
+              origin:
+                default: multikueue
+                description: A label value used to track the creator of workloads
+                  in the worker cluster.
+                maxLength: 63
+                type: string
+                x-kubernetes-validations:
+                - message: Value is immutable
+                  rule: self == oldSelf
             required:
             - kubeConfig
             type: object

diff --git a/client-go/applyconfiguration/kueue/v1alpha1/multikueueclusterspec.go b/client-go/applyconfiguration/kueue/v1alpha1/multikueueclusterspec.go
diff --git a/cmd/kueue/main.go b/cmd/kueue/main.go
@@ -244,7 +244,7 @@ func setupControllers(mgr ctrl.Manager, cCache *cache.Cache, queues *queue.Manag
 	}
 
 	if features.Enabled(features.MultiKueue) {
-		if err := multikueue.SetupControllers(mgr, *cfg.Namespace); err != nil {
+		if err := multikueue.SetupControllers(mgr, *cfg.Namespace, multikueue.WithGCInterval(cfg.MultiKueue.GCTimeout.Duration)); err != nil {
 			setupLog.Error(err, "Could not setup MultiKueue controller")
 			os.Exit(1)
 		}

diff --git a/cmd/kueue/main_test.go b/cmd/kueue/main_test.go
@@ -117,6 +117,7 @@ integrations:
 						MaxCount: config.DefaultClusterQueuesMaxCount,
 					},
 				},
+				MultiKueue: &config.MultiKueue{GCTimeout: &metav1.Duration{Duration: config.DefaultMultiKueueGCTimeout}},
 			},
 		},
 		{

diff --git a/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml b/config/components/crd/bases/kueue.x-k8s.io_multikueueclusters.yaml
@@ -52,6 +52,15 @@ spec:
                 - location
                 - locationType
                 type: object
+              origin:
+                default: multikueue
+                description: A label value used to track the creator of workloads
+                  in the worker cluster.
+                maxLength: 63
+                type: string
+                x-kubernetes-validations:
+                - message: Value is immutable
+                  rule: self == oldSelf
             required:
             - kubeConfig
             type: object