[WIP] Reclaim backoff for re-borrowing after preemption-driven reclamation

apis/config/v1beta2/configuration_types.go

@@ -112,6 +112,14 @@ type Configuration struct {
 	// VisibilityServer configures the visibility server.
 	// +optional
 	VisibilityServer *VisibilityServerConfiguration `json:"visibilityServer,omitempty"`
+
+	// ReclaimBackoff configures the per-resource reclaim backoff. After a
+	// ClusterQueue's borrowed resource is reclaimed by preemption, the scheduler
+	// applies an exponential cooldown that defers only the assignments which would
+	// borrow that same resource again. The feature is enabled only when this field
+	// is set and its Enable subfield is true; unset or Enable=false disables it.
+	// +optional
+	ReclaimBackoff *ReclaimBackoff `json:"reclaimBackoff,omitempty"`
 }
 
 type ControllerManager struct {
@@ -329,6 +337,52 @@ type WaitForPodsReady struct {
 	RecoveryTimeout *metav1.Duration `json:"recoveryTimeout,omitempty"`
 }
 
+// ReclaimBackoff defines configuration for the per-resource reclaim backoff.
+// After a ClusterQueue's borrowed resource is reclaimed by preemption, the
+// scheduler defers, for a cooldown window, only the assignments that would
+// borrow that same resource again on the same ClusterQueue. Assignments that
+// fit within nominal quota, and assignments of other resources, are unaffected.
+// The feature is enabled only when Enable is true; otherwise the scheduler
+// behaves as if the feature is off.
+type ReclaimBackoff struct {
+	// Enable controls whether the per-resource reclaim backoff is active. It
+	// must be set to true to turn the feature on; when unset or false, the
+	// remaining fields in this struct are ignored and the scheduler behaves as
+	// if the feature is disabled.
+	// +optional
+	Enable *bool `json:"enable,omitempty"`
+
+	// BackoffBaseSeconds defines the base for the exponential backoff applied to
+	// a (ClusterQueue, resource) pair after its borrowed quota is reclaimed.
+	//
+	// The cooldown for the n-th consecutive reclaim is about "b*2^(n-1)+Rand"
+	// where "b" is BackoffBaseSeconds and "Rand" is a small random jitter, capped
+	// at BackoffMaxSeconds. By default, the consecutive cooldowns are around
+	// (60s, 120s, 240s, ...).
+	//
+	// Defaults to 60.
+	// +optional
+	BackoffBaseSeconds *int32 `json:"backoffBaseSeconds,omitempty"`
+
+	// BackoffMaxSeconds defines the maximum cooldown, in seconds, applied to a
+	// single (ClusterQueue, resource) pair.
+	//
+	// Defaults to 3600.
+	// +optional
+	BackoffMaxSeconds *int32 `json:"backoffMaxSeconds,omitempty"`
+
+	// BackoffResetSeconds defines the quiet period, in seconds, after which the
+	// consecutive-reclaim counter for a (ClusterQueue, resource) pair is reset. If
+	// the pair is not reclaimed again within this period, the next reclaim starts
+	// the backoff from BackoffBaseSeconds. This value should be noticeably larger
+	// than BackoffBaseSeconds; otherwise the counter resets within a single base
+	// window and the backoff never grows.
+	//
+	// Defaults to 600.
+	// +optional
+	BackoffResetSeconds *int32 `json:"backoffResetSeconds,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.

apis/config/v1beta2/defaults.go

@@ -50,6 +50,9 @@ const (
 	DefaultMultiKueueWorkerLostTimeout            = 15 * time.Minute
 	DefaultRequeuingBackoffBaseSeconds            = 60
 	DefaultRequeuingBackoffMaxSeconds             = 3600
+	DefaultReclaimBackoffBaseSeconds              = 60
+	DefaultReclaimBackoffMaxSeconds               = 3600
+	DefaultReclaimBackoffResetSeconds             = 600
 	DefaultResourceTransformationStrategy         = Retain
 	DefaultVisibilityBindPort                     = 8082
 	DefaultCustomMetricLabelSourceKind            = SourceKindClusterQueue
@@ -116,6 +119,12 @@ func SetDefaults_Configuration(cfg *Configuration) {
 		cfg.WaitForPodsReady.RequeuingStrategy.BackoffMaxSeconds = cmp.Or(cfg.WaitForPodsReady.RequeuingStrategy.BackoffMaxSeconds, new(int32(DefaultRequeuingBackoffMaxSeconds)))
 	}
 
+	// ReclaimBackoff is intentionally not defaulted here: the feature is opt-in
+	// via the explicit Enable=true field, and the scheduler applies the per-field
+	// defaults (DefaultReclaimBackoff*) when it builds the tracker. Defaulting
+	// the block or Enable here would silently turn the feature on for every
+	// installation.
+
 	cfg.Integrations = cmp.Or(cfg.Integrations, &Integrations{})
 	if len(cfg.Integrations.Frameworks) == 0 {
 		cfg.Integrations.Frameworks = []string{defaultJobFrameworkName}

apis/kueue/v1beta2/workload_types.go

@@ -978,6 +978,13 @@ const (
 	// for previously admitted workloads to reach PodsReady condition under waitForPodsReady configuration.
 	WorkloadQuotaReservedReasonWaitingForPodsReady = "WaitingForPodsReady"
 
+	// WorkloadQuotaReservedReasonReclaimBackoff indicates that the workload is waiting because
+	// its ClusterQueue recently had a borrowed resource reclaimed by preemption, and the
+	// resource it would borrow is in the reclaim backoff cooldown. Only assignments that would
+	// borrow the reclaimed resource are deferred; assignments within nominal quota, and other
+	// resources, are unaffected. Only reported when Configuration.ReclaimBackoff is set.
+	WorkloadQuotaReservedReasonReclaimBackoff = "ReclaimBackoff"
+
 	// WorkloadAdmittedReasonNoReservation indicates that the workload has no reservation.
 	WorkloadAdmittedReasonNoReservation = "NoReservation"
 

cmd/kueue/main.go

@@ -25,6 +25,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"time"
 
 	zaplog "go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
@@ -39,6 +40,7 @@ import (
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
 	"k8s.io/client-go/rest"
 	"k8s.io/client-go/util/flowcontrol"
+	"k8s.io/utils/clock"
 	"k8s.io/utils/ptr"
 	inventoryv1alpha1 "sigs.k8s.io/cluster-inventory-api/apis/v1alpha1"
 	ctrl "sigs.k8s.io/controller-runtime"
@@ -79,6 +81,7 @@ import (
 	"sigs.k8s.io/kueue/pkg/scheduler"
 	preemptexpectations "sigs.k8s.io/kueue/pkg/scheduler/preemption/expectations"
 	"sigs.k8s.io/kueue/pkg/scheduler/preemption/fairsharing"
+	"sigs.k8s.io/kueue/pkg/scheduler/reclaimbackoff"
 	"sigs.k8s.io/kueue/pkg/util/cert"
 	utildra "sigs.k8s.io/kueue/pkg/util/dra"
 	"sigs.k8s.io/kueue/pkg/util/expectations"
@@ -416,6 +419,7 @@ func main() {
 		DRABackedResources:        draBackedResources,
 		ResourceFormatter:         resourceFormatter,
 		ResourceSliceAPIAvailable: resourceSliceAPIAvailable,
+		ReclaimBackoff:            reclaimBackoffTracker(&cfg),
 	}
 	if err := setupControllers(ctx, mgr, cCache, queues, &cfg, serverVersionFetcher, integrationManager, controllerOpts); err != nil {
 		setupLog.Error(err, "Unable to setup controllers")
@@ -439,7 +443,9 @@ func main() {
 		}()
 	}
 
-	if err := setupScheduler(mgr, cCache, queues, &cfg, roleTracker, preemptionExpectations, customLabels, resourceFormatter); err != nil {
+	// The tracker instance must be shared with the ClusterQueue reconciler so
+	// that deleting a ClusterQueue purges the entries the scheduler armed for it.
+	if err := setupScheduler(mgr, cCache, queues, &cfg, roleTracker, preemptionExpectations, customLabels, resourceFormatter, controllerOpts.ReclaimBackoff); err != nil {
 		setupLog.Error(err, "Could not setup scheduler")
 		os.Exit(1)
 	}
@@ -671,6 +677,7 @@ func setupScheduler(
 	preemptionExpectations *expectations.Store,
 	customLabels *metrics.CustomLabels,
 	resourceFormatter *resources.ResourceFormatter,
+	reclaimBackoff *reclaimbackoff.Tracker,
 ) error {
 	sched := scheduler.New(
 		queues,
@@ -685,13 +692,31 @@ func setupScheduler(
 		scheduler.WithPreemptionExpectations(preemptionExpectations),
 		scheduler.WithCustomLabels(customLabels),
 		scheduler.WithResourceFormatter(resourceFormatter),
+		scheduler.WithReclaimBackoff(reclaimBackoff),
 	)
 	if err := mgr.Add(sched); err != nil {
 		return fmt.Errorf("unable to add scheduler to manager: %w", err)
 	}
 	return nil
 }
 
+// reclaimBackoffTracker builds the reclaim backoff tracker from cfg, or returns
+// nil when the reclaimBackoff block is missing or its Enable field is not true,
+// so the scheduler treats the feature as off. Individual fields fall back to
+// DefaultReclaimBackoff*.
+func reclaimBackoffTracker(cfg *configapi.Configuration) *reclaimbackoff.Tracker {
+	if cfg.ReclaimBackoff == nil || !ptr.Deref(cfg.ReclaimBackoff.Enable, false) {
+		return nil
+	}
+	rb := cfg.ReclaimBackoff
+	return reclaimbackoff.New(
+		time.Duration(ptr.Deref(rb.BackoffBaseSeconds, configapi.DefaultReclaimBackoffBaseSeconds))*time.Second,
+		time.Duration(ptr.Deref(rb.BackoffMaxSeconds, configapi.DefaultReclaimBackoffMaxSeconds))*time.Second,
+		time.Duration(ptr.Deref(rb.BackoffResetSeconds, configapi.DefaultReclaimBackoffResetSeconds))*time.Second,
+		clock.RealClock{},
+	)
+}
+
 func setupServerVersionFetcher(mgr ctrl.Manager, kubeConfig *rest.Config) (*kubeversion.ServerVersionFetcher, error) {
 	discoveryClient, err := discovery.NewDiscoveryClientForConfig(kubeConfig)
 	if err != nil {

pkg/cache/queue/inadmissible_workloads.go

@@ -204,6 +204,28 @@ func notifyRetryInadmissibleWithoutLock(m *Manager, cqNames sets.Set[kueue.Clust
 	}
 }
 
+// NotifyRetryInadmissibleAfter requests, after at least delay has elapsed, that
+// inadmissible workloads from the given ClusterQueue (and its whole Cohort tree)
+// be moved from the inadmissible queue back to the active heap. It is used to
+// wake a ClusterQueue once a time-based gate such as reclaim backoff is expected
+// to have expired, since ordinary retries are only triggered by quota-freeing
+// events, which may not occur while the Cohort is idle.
+func (m *Manager) NotifyRetryInadmissibleAfter(cqName kueue.ClusterQueueReference, delay time.Duration) {
+	m.RLock()
+	defer m.RUnlock()
+	cq := m.hm.ClusterQueue(cqName)
+	if cq == nil {
+		return
+	}
+	switch {
+	case !cq.HasParent():
+		m.requeuer.notifyClusterQueueAfter(cq.name, delay)
+	case !hierarchy.HasCycle(cq.Parent()):
+		rootName := cq.Parent().getRootUnsafe().GetName()
+		m.requeuer.notifyCohortAfter(rootName, delay)
+	}
+}
+
 // inadmissibleRequeuer receives notifications
 // that a particular ClusterQueue (without Cohort) or a
 // Root Cohort should have its Inadmissible Workloads requeued.
@@ -212,6 +234,13 @@ type inadmissibleRequeuer interface {
 	notifyClusterQueue(cqName kueue.ClusterQueueReference)
 	// notifyCohort should only be called for Root Cohorts.
 	notifyCohort(cohortName kueue.CohortReference)
+	// notifyClusterQueueAfter behaves like notifyClusterQueue but defers the
+	// requeue by at least delay. It is used to wake a ClusterQueue once a
+	// time-based gate (e.g. reclaim backoff) is expected to have expired.
+	notifyClusterQueueAfter(cqName kueue.ClusterQueueReference, delay time.Duration)
+	// notifyCohortAfter behaves like notifyCohort but defers the requeue by at
+	// least delay.
+	notifyCohortAfter(cohortName kueue.CohortReference, delay time.Duration)
 	setManager(manager *Manager)
 }
 
@@ -249,6 +278,14 @@ func (r *workqueueRequeuer) notifyCohort(cohortName kueue.CohortReference) {
 	r.queue.AddAfter(requeueRequest{Cohort: cohortName}, r.batchPeriod)
 }
 
+func (r *workqueueRequeuer) notifyClusterQueueAfter(cqName kueue.ClusterQueueReference, delay time.Duration) {
+	r.queue.AddAfter(requeueRequest{ClusterQueue: cqName}, max(delay, r.batchPeriod))
+}
+
+func (r *workqueueRequeuer) notifyCohortAfter(cohortName kueue.CohortReference, delay time.Duration) {
+	r.queue.AddAfter(requeueRequest{Cohort: cohortName}, max(delay, r.batchPeriod))
+}
+
 func (r *workqueueRequeuer) setManager(manager *Manager) {
 	r.manager = manager
 }

pkg/cache/queue/test_util.go

@@ -18,6 +18,7 @@ package queue
 
 import (
 	"context"
+	"time"
 
 	"k8s.io/apimachinery/pkg/util/sets"
 	"sigs.k8s.io/controller-runtime/pkg/client"
@@ -41,6 +42,18 @@ func (r *testInadmissibleWorkloadRequeuer) notifyCohort(cohortName kueue.CohortR
 	r.cohorts.Insert(cohortName)
 }
 
+// notifyClusterQueueAfter buffers the requeue like notifyClusterQueue; the delay
+// is ignored because ProcessRequeues is driven synchronously by tests.
+func (r *testInadmissibleWorkloadRequeuer) notifyClusterQueueAfter(cqName kueue.ClusterQueueReference, _ time.Duration) {
+	r.cqs.Insert(cqName)
+}
+
+// notifyCohortAfter buffers the requeue like notifyCohort; the delay is ignored
+// because ProcessRequeues is driven synchronously by tests.
+func (r *testInadmissibleWorkloadRequeuer) notifyCohortAfter(cohortName kueue.CohortReference, _ time.Duration) {
+	r.cohorts.Insert(cohortName)
+}
+
 func (r *testInadmissibleWorkloadRequeuer) setManager(manager *Manager) {
 	r.manager = manager
 }