[multikueue] Add garbage collector.

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,21 @@ type WaitForPodsReady struct {
 	RequeuingTimestamp *RequeuingTimestamp `json:"requeuingTimestamp,omitempty"`
 }
 
+type MultiKueue struct {
+	// GCInterval defines the time interval between two consecutive garbage collection runs.
+	// Defaults to 1min. If 0, the garbage collection is disabled.
+	// +optional
+	GCInterval *metav1.Duration `json:"gcInterval"`
+
+	// Origin defines a label value used to track the creator of workloads in the worker
+	// clusters.
+	// This is used by multikueue in components like its garbage collector to identify
+	// remote objects that ware created by this multikueue manager cluster and delete
+	// them if their local counterpart no longer exists.
+	// +optional
+	Origin *string `json:"origin,omitempty"`
+}
+
 type RequeuingTimestamp string
 
 const (

apis/config/v1beta1/defaults.go

@@ -44,6 +44,8 @@ const (
 	DefaultQueueVisibilityUpdateIntervalSeconds int32   = 5
 	DefaultClusterQueuesMaxCount                int32   = 10
 	defaultJobFrameworkName                             = "batch/job"
+	DefaultMultiKueueGCInterval                         = time.Minute
+	DefaultMultiKueueOrigin                             = "multikueue"
 )
 
 func getOperatorNamespace() string {
@@ -161,4 +163,14 @@ 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.GCInterval == nil {
+		cfg.MultiKueue.GCInterval = &metav1.Duration{Duration: DefaultMultiKueueGCInterval}
+	}
+	if ptr.Deref(cfg.MultiKueue.Origin, "") == "" {
+		cfg.MultiKueue.Origin = ptr.To(DefaultMultiKueueOrigin)
+	}
 }

apis/config/v1beta1/defaults_test.go

@@ -96,6 +96,11 @@ func TestSetDefaults_Configuration(t *testing.T) {
 		},
 	}
 
+	defaultMultiKueue := &MultiKueue{
+		GCInterval: &metav1.Duration{Duration: DefaultMultiKueueGCInterval},
+		Origin:     ptr.To(DefaultMultiKueueOrigin),
+	}
+
 	podsReadyTimeoutTimeout := metav1.Duration{Duration: defaultPodsReadyTimeout}
 	podsReadyTimeoutOverwrite := metav1.Duration{Duration: time.Minute}
 
@@ -118,6 +123,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting ControllerManager": {
@@ -158,6 +164,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default ControllerManager": {
@@ -214,6 +221,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not set LeaderElectionID": {
@@ -254,6 +262,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting InternalCertManagement": {
@@ -271,6 +280,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default InternalCertManagement": {
@@ -289,6 +299,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"should not default values in custom ClientConnection": {
@@ -314,6 +325,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				},
 				Integrations:    overwriteNamespaceIntegrations,
 				QueueVisibility: defaultQueueVisibility,
+				MultiKueue:      defaultMultiKueue,
 			},
 		},
 		"should default empty custom ClientConnection": {
@@ -333,6 +345,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     overwriteNamespaceIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"defaulting waitForPodsReady values": {
@@ -359,6 +372,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 +399,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"respecting provided waitForPodsReady values": {
@@ -413,6 +428,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 				ClientConnection: defaultClientConnection,
 				Integrations:     defaultIntegrations,
 				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue:       defaultMultiKueue,
 			},
 		},
 		"integrations": {
@@ -436,6 +452,7 @@ func TestSetDefaults_Configuration(t *testing.T) {
 					PodOptions: defaultIntegrations.PodOptions,
 				},
 				QueueVisibility: defaultQueueVisibility,
+				MultiKueue:      defaultMultiKueue,
 			},
 		},
 		"queue visibility": {
@@ -464,6 +481,57 @@ func TestSetDefaults_Configuration(t *testing.T) {
 						MaxCount: 0,
 					},
 				},
+				MultiKueue: defaultMultiKueue,
+			},
+		},
+		"multiKueue": {
+			original: &Configuration{
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				MultiKueue: &MultiKueue{
+					GCInterval: &metav1.Duration{Duration: time.Second},
+					Origin:     ptr.To("multikueue-manager1"),
+				},
+			},
+			want: &Configuration{
+				Namespace:         ptr.To(DefaultNamespace),
+				ControllerManager: defaultCtrlManagerConfigurationSpec,
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				ClientConnection: defaultClientConnection,
+				Integrations:     defaultIntegrations,
+				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue: &MultiKueue{
+					GCInterval: &metav1.Duration{Duration: time.Second},
+					Origin:     ptr.To("multikueue-manager1"),
+				},
+			},
+		},
+		"multiKueue GCInterval 0": {
+			original: &Configuration{
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				MultiKueue: &MultiKueue{
+					GCInterval: &metav1.Duration{},
+					Origin:     ptr.To("multikueue-manager1"),
+				},
+			},
+			want: &Configuration{
+				Namespace:         ptr.To(DefaultNamespace),
+				ControllerManager: defaultCtrlManagerConfigurationSpec,
+				InternalCertManagement: &InternalCertManagement{
+					Enable: ptr.To(false),
+				},
+				ClientConnection: defaultClientConnection,
+				Integrations:     defaultIntegrations,
+				QueueVisibility:  defaultQueueVisibility,
+				MultiKueue: &MultiKueue{
+					GCInterval: &metav1.Duration{},
+					Origin:     ptr.To("multikueue-manager1"),
+				},
 			},
 		},
 	}

apis/kueue/v1alpha1/multikueue_types.go

@@ -23,6 +23,10 @@ import (
 const (
 	MultiKueueConfigSecretKey = "kubeconfig"
 	MultiKueueClusterActive   = "Active"
+
+	// MultiKueueOriginLabel is a label used to track the creator
+	// of multikueue remote objects.
+	MultiKueueOriginLabel = "kueue.x-k8s.io/multikueue-origin"
 )
 
 type LocationType string

cmd/kueue/main.go

@@ -38,6 +38,7 @@ import (
 	"k8s.io/client-go/discovery"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
+	"k8s.io/utils/ptr"
 	ctrl "sigs.k8s.io/controller-runtime"
 	"sigs.k8s.io/controller-runtime/pkg/healthz"
 	"sigs.k8s.io/controller-runtime/pkg/log/zap"
@@ -245,7 +246,10 @@ 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.GCInterval.Duration),
+			multikueue.WithOrigin(ptr.Deref(cfg.MultiKueue.Origin, configapi.DefaultMultiKueueOrigin)),
+		); err != nil {
 			setupLog.Error(err, "Could not setup MultiKueue controller")
 			os.Exit(1)
 		}

cmd/kueue/main_test.go

@@ -117,6 +117,10 @@ integrations:
 						MaxCount: config.DefaultClusterQueuesMaxCount,
 					},
 				},
+				MultiKueue: &config.MultiKueue{
+					GCInterval: &metav1.Duration{Duration: config.DefaultMultiKueueGCInterval},
+					Origin:     ptr.To(config.DefaultMultiKueueOrigin),
+				},
 			},
 		},
 		{