[jobframework] Deployment integration

apis/config/v1beta1/configuration_types.go

@@ -318,6 +318,7 @@ type Integrations struct {
 	//  - "kubeflow.org/tfjob"
 	//  - "kubeflow.org/xgboostjob"
 	//  - "pod"
+	//  - "deployment"
 	Frameworks []string `json:"frameworks,omitempty"`
 	// List of GroupVersionKinds that are managed for Kueue by external controllers;
 	// the expected format is `Kind.version.group.com`.
@@ -344,6 +345,13 @@ type PodIntegrationOptions struct {
 	PodSelector *metav1.LabelSelector `json:"podSelector,omitempty"`
 }
 
+type DeploymentIntegrationOptions struct {
+	// NamespaceSelector can be used to omit some namespaces from pod reconciliation
+	NamespaceSelector *metav1.LabelSelector `json:"namespaceSelector,omitempty"`
+	// DeploymentSelector can be used to choose what deployments to reconcile
+	DeploymentSelector *metav1.LabelSelector `json:"deploymentSelector,omitempty"`
+}
+
 type QueueVisibility struct {
 	// ClusterQueues is configuration to expose the information
 	// about the top pending workloads in the cluster queue.

pkg/controller/jobs/deployment/deployment_controller.go

@@ -0,0 +1,68 @@
+/*
+Copyright 2024 The Kubernetes Authors.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package deployment
+
+import (
+	"context"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/apimachinery/pkg/runtime/schema"
+	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+
+	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+	"sigs.k8s.io/kueue/pkg/controller/jobs/noop"
+)
+
+var (
+	gvk = corev1.SchemeGroupVersion.WithKind("Deployment")
+)
+
+const (
+	FrameworkName = "deployment"
+)
+
+func init() {
+	utilruntime.Must(jobframework.RegisterIntegration(FrameworkName, jobframework.IntegrationCallbacks{
+		SetupIndexes:   SetupIndexes,
+		NewReconciler:  noop.NewReconciler,
+		SetupWebhook:   SetupWebhook,
+		JobType:        &appsv1.Deployment{},
+		AddToScheme:    appsv1.AddToScheme,
+		DependencyList: []string{"pod"},
+	}))
+}
+
+type Deployment appsv1.Deployment
+
+func FromObject(o runtime.Object) *Deployment {
+	return (*Deployment)(o.(*appsv1.Deployment))
+}
+
+func (d *Deployment) Object() client.Object {
+	return (*appsv1.Deployment)(d)
+}
+
+func (d *Deployment) GVK() schema.GroupVersionKind {
+	return gvk
+}
+
+func SetupIndexes(ctx context.Context, indexer client.FieldIndexer) error {
+	return jobframework.SetupWorkloadOwnerIndex(ctx, indexer, gvk)
+}

pkg/controller/jobs/deployment/deployment_webhook.go

@@ -0,0 +1,108 @@
+package deployment
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	appsv1 "k8s.io/api/apps/v1"
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/labels"
+	"k8s.io/apimachinery/pkg/runtime"
+	"k8s.io/klog/v2"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/webhook"
+
+	configapi "sigs.k8s.io/kueue/apis/config/v1beta1"
+	"sigs.k8s.io/kueue/pkg/controller/constants"
+	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+)
+
+var (
+	errDeploymentOptsTypeAssertion = errors.New("options are not of type DeploymentIntegrationOptions")
+	errDeploymentOptsNotFound      = errors.New("deploymentIntegrationOptions not found in options")
+)
+
+type Webhook struct {
+	client                     client.Client
+	manageJobsWithoutQueueName bool
+	namespaceSelector          *metav1.LabelSelector
+	deploymentSelector         *metav1.LabelSelector
+}
+
+func SetupWebhook(mgr ctrl.Manager, opts ...jobframework.Option) error {
+	options := jobframework.ProcessOptions(opts...)
+	deploymentOpts, err := getDeploymentOptions(options.IntegrationOptions)
+	if err != nil {
+		return err
+	}
+	wh := &Webhook{
+		client:                     mgr.GetClient(),
+		manageJobsWithoutQueueName: options.ManageJobsWithoutQueueName,
+		namespaceSelector:          deploymentOpts.NamespaceSelector,
+		deploymentSelector:         deploymentOpts.DeploymentSelector,
+	}
+	return ctrl.NewWebhookManagedBy(mgr).
+		For(&appsv1.Deployment{}).
+		WithDefaulter(wh).
+		Complete()
+}
+
+func getDeploymentOptions(integrationOpts map[string]any) (configapi.DeploymentIntegrationOptions, error) {
+	opts, ok := integrationOpts[corev1.SchemeGroupVersion.WithKind("Deployment").String()]
+	if !ok {
+		return configapi.DeploymentIntegrationOptions{}, errDeploymentOptsNotFound
+	}
+	deploymentOpts, ok := opts.(*configapi.DeploymentIntegrationOptions)
+	if !ok {
+		return configapi.DeploymentIntegrationOptions{}, fmt.Errorf("%w, got %T", errDeploymentOptsTypeAssertion, opts)
+	}
+	return *deploymentOpts, nil
+}
+
+// +kubebuilder:webhook:path=/mutate--v1-deployment,mutating=true,failurePolicy=fail,sideEffects=None,groups="",resources=pods,verbs=create,versions=v1,name=mpod.kb.io,admissionReviewVersions=v1
+// +kubebuilder:rbac:groups="",resources=namespaces,verbs=get;list;watch
+
+var _ webhook.CustomDefaulter = &Webhook{}
+
+func (wh *Webhook) Default(ctx context.Context, obj runtime.Object) error {
+	d := FromObject(obj)
+	log := ctrl.LoggerFrom(ctx).WithName("deployment-webhook").WithValues("deployment", klog.KObj(d))
+	log.V(5).Info("Applying defaults")
+
+	// Check for deployment label selector match
+	deploymentSelector, err := metav1.LabelSelectorAsSelector(wh.deploymentSelector)
+	if err != nil {
+		return fmt.Errorf("failed to create deployment selector: %w", err)
+	}
+	if !deploymentSelector.Matches(labels.Set(d.Labels)) {
+		return nil
+	}
+
+	// Get deployment namespace and check for namespace label selector match
+	ns := &corev1.Namespace{}
+	if err := wh.client.Get(ctx, client.ObjectKey{Name: d.Namespace}, ns); err != nil {
+		return fmt.Errorf("failed to run mutating webhook on deployment %s/%s, error while getting namespace: %w",
+			d.Namespace,
+			d.Name,
+			err,
+		)
+	}
+	log.V(5).Info("Found deployment namespace", "Namespace.Name", ns.GetName())
+	nsSelector, err := metav1.LabelSelectorAsSelector(wh.namespaceSelector)
+	if err != nil {
+		return fmt.Errorf("failed to parse namespace selector: %w", err)
+	}
+	if !nsSelector.Matches(labels.Set(ns.GetLabels())) {
+		return nil
+	}
+
+	if d.Spec.Template.Labels == nil {
+		d.Spec.Template.Labels = make(map[string]string)
+	}
+	d.Spec.Template.Labels[constants.QueueLabel] = d.Labels[constants.QueueLabel]
+
+	return nil
+}

pkg/controller/jobs/noop/noop_controller.go

@@ -0,0 +1,40 @@
+package noop
+
+import (
+	"context"
+
+	corev1 "k8s.io/api/core/v1"
+	"k8s.io/client-go/tools/record"
+	ctrl "sigs.k8s.io/controller-runtime"
+	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/controller"
+	"sigs.k8s.io/controller-runtime/pkg/reconcile"
+
+	"sigs.k8s.io/kueue/pkg/controller/jobframework"
+)
+
+var (
+	_   jobframework.JobReconcilerInterface = (*Reconciler)(nil)
+	gvk                                     = corev1.SchemeGroupVersion.WithKind("Noop")
+)
+
+type Reconciler struct {
+}
+
+func (r Reconciler) Reconcile(_ context.Context, _ reconcile.Request) (reconcile.Result, error) {
+	return ctrl.Result{}, nil
+}
+
+func (r Reconciler) SetupWithManager(mgr ctrl.Manager) error {
+	concurrency := mgr.GetControllerOptions().GroupKindConcurrency[gvk.GroupKind().String()]
+	ctrl.Log.V(3).Info("Setting up Noop reconciler", "concurrency", concurrency)
+	return ctrl.NewControllerManagedBy(mgr).
+		WithOptions(controller.Options{
+			MaxConcurrentReconciles: concurrency,
+		}).
+		Complete(r)
+}
+
+func NewReconciler(client client.Client, record record.EventRecorder, opts ...jobframework.Option) jobframework.JobReconcilerInterface {
+	return &Reconciler{}
+}