feat: CR safeguards, throttle CRD creation, adding limit

cmd/manager/main.go

@@ -581,19 +581,20 @@ func main() {
 		}
 
 		crControllerConf := commitmentsConfig.CommittedResourceController
-		crControllerConf.ApplyDefaults()
+
+		crControllerMonitor := commitments.NewCRControllerMonitor(multiclusterClient)
+		metrics.Registry.MustRegister(&crControllerMonitor)
+
 		if err := (&commitments.CommittedResourceController{
-			Client: multiclusterClient,
-			Scheme: mgr.GetScheme(),
-			Conf:   crControllerConf,
+			Client:  multiclusterClient,
+			Scheme:  mgr.GetScheme(),
+			Conf:    crControllerConf,
+			Monitor: &crControllerMonitor,
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "CommittedResource")
 			os.Exit(1)
 		}
 
-		crControllerMonitor := commitments.NewCRControllerMonitor(multiclusterClient)
-		metrics.Registry.MustRegister(&crControllerMonitor)
-
 		usageReconcilerMonitor := commitments.NewUsageReconcilerMonitor()
 		metrics.Registry.MustRegister(&usageReconcilerMonitor)
 		if commitmentsUsageDB == nil {

helm/bundles/cortex-nova/values.yaml

@@ -173,6 +173,10 @@ cortex-scheduling-controllers:
       requeueIntervalRetry: "1m"
       # Maximum back-off interval cap for the exponential retry delay
       maxRequeueInterval: "30m"
+      # Pause between consecutive Reservation CRD creates to spread scheduler load; 0 disables
+      slotCreationDelay: "0ms"
+      # Max Reservation CRDs per CommittedResource on the API path; 0 disables the limit
+      maxSlotsPerCommitment: 0
     committedResourceAPI:
       # Timeout for watching CommittedResource CRDs before rolling back
       watchTimeout: "15s"

internal/scheduling/reservations/commitments/committed_resource_controller.go

@@ -5,6 +5,7 @@ package commitments
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"time"
 
@@ -34,8 +35,9 @@ const (
 // CommittedResourceController reconciles CommittedResource CRDs and owns all child Reservation CRUD.
 type CommittedResourceController struct {
 	client.Client
-	Scheme *runtime.Scheme
-	Conf   CommittedResourceControllerConfig
+	Scheme  *runtime.Scheme
+	Conf    CommittedResourceControllerConfig
+	Monitor *CRControllerMonitor
 }
 
 func (r *CommittedResourceController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) {
@@ -313,8 +315,17 @@ func (r *CommittedResourceController) applyReservationState(ctx context.Context,
 	state.CreatorRequestID = reservations.GlobalRequestIDFromContext(ctx)
 	state.ParentGeneration = cr.Generation
 
-	result, err := NewReservationManager(r.Client).ApplyCommitmentState(ctx, logger, state, flavorGroups, "committed-resource-controller")
+	mgr := NewReservationManager(r.Client)
+	mgr.SlotCreationDelay = r.Conf.SlotCreationDelay.Duration
+	if cr.Spec.AllowRejection {
+		mgr.MaxSlots = r.Conf.MaxSlotsPerCommitment
+	}
+	result, err := mgr.ApplyCommitmentState(ctx, logger, state, flavorGroups, "committed-resource-controller")
 	if err != nil {
+		var limitErr *SlotLimitExceededError
+		if errors.As(err, &limitErr) && r.Monitor != nil {
+			r.Monitor.RecordSlotLimitRejection(cr.Spec.FlavorGroupName, cr.Spec.AvailabilityZone)
+		}
 		return nil, err
 	}
 	logger.Info("commitment state applied", "created", result.Created, "deleted", result.Deleted, "repaired", result.Repaired)

internal/scheduling/reservations/commitments/committed_resource_controller_monitor.go

@@ -23,8 +23,9 @@ var crControllerMonitorLog = ctrl.Log.WithName("committed-resource-controller-mo
 // after a failure. API-originated dry-run probes (AllowRejection=true) are excluded.
 // The metric is absent for a given label set when no CRs match — absence means zero.
 type CRControllerMonitor struct {
-	client      client.Client
-	unfulfilled *prometheus.Desc
+	client              client.Client
+	unfulfilled         *prometheus.Desc
+	slotLimitRejections *prometheus.CounterVec
 }
 
 func NewCRControllerMonitor(c client.Client) CRControllerMonitor {
@@ -36,12 +37,22 @@ func NewCRControllerMonitor(c client.Client) CRControllerMonitor {
 			[]string{"flavor_group", "resource_type", "availability_zone"},
 			nil,
 		),
+		slotLimitRejections: prometheus.NewCounterVec(prometheus.CounterOpts{
+			Name: "cortex_committed_resource_slot_limit_rejections_total",
+			Help: "Number of times a commitment was rejected because the requested slot count exceeded the configured limit.",
+		}, []string{"flavor_group", "availability_zone"}),
 	}
 }
 
+// RecordSlotLimitRejection increments the slot-limit rejection counter for the given flavor group and AZ.
+func (m *CRControllerMonitor) RecordSlotLimitRejection(flavorGroup, az string) {
+	m.slotLimitRejections.WithLabelValues(flavorGroup, az).Inc()
+}
+
 // Describe implements prometheus.Collector.
 func (m *CRControllerMonitor) Describe(ch chan<- *prometheus.Desc) {
 	ch <- m.unfulfilled
+	m.slotLimitRejections.Describe(ch)
 }
 
 // Collect implements prometheus.Collector. Lists all CommittedResource CRDs and counts
@@ -80,4 +91,5 @@ func (m *CRControllerMonitor) Collect(ch chan<- prometheus.Metric) {
 			k.flavorGroup, k.resourceType, k.az,
 		)
 	}
+	m.slotLimitRejections.Collect(ch)
 }

internal/scheduling/reservations/commitments/config.go

@@ -79,24 +79,18 @@ type CommittedResourceControllerConfig struct {
 	// MaxRequeueInterval caps the exponential backoff delay.
 	// Once this ceiling is reached, every subsequent retry fires after exactly this interval.
 	MaxRequeueInterval metav1.Duration `json:"maxRequeueInterval"`
-}
-
-func DefaultCommittedResourceControllerConfig() CommittedResourceControllerConfig {
-	return CommittedResourceControllerConfig{
-		RequeueIntervalRetry: metav1.Duration{Duration: 30 * time.Second},
-		MaxRequeueInterval:   metav1.Duration{Duration: 30 * time.Minute},
-	}
-}
 
-// ApplyDefaults fills in zero-value fields from the defaults, leaving explicitly configured values intact.
-func (c *CommittedResourceControllerConfig) ApplyDefaults() {
-	d := DefaultCommittedResourceControllerConfig()
-	if c.RequeueIntervalRetry.Duration == 0 {
-		c.RequeueIntervalRetry = d.RequeueIntervalRetry
-	}
-	if c.MaxRequeueInterval.Duration == 0 {
-		c.MaxRequeueInterval = d.MaxRequeueInterval
-	}
+	// SlotCreationDelay is the pause inserted between consecutive Reservation CRD creates.
+	// Spreads scheduler calls over time instead of bursting them all at once.
+	// 0 disables the delay.
+	SlotCreationDelay metav1.Duration `json:"slotCreationDelay"`
+
+	// MaxSlotsPerCommitment caps the number of Reservation CRDs that may be created for a single
+	// CommittedResource on the AllowRejection=true (API) path. Requests that would exceed this
+	// limit are rejected immediately before any slots are created.
+	// Has no effect on the AllowRejection=false (syncer) path.
+	// 0 disables the cap.
+	MaxSlotsPerCommitment int `json:"maxSlotsPerCommitment"`
 }
 
 // ResourceTypeConfig holds per-resource flags for a single resource type within a flavor group.

internal/scheduling/reservations/commitments/reservation_manager.go

@@ -6,6 +6,7 @@ package commitments
 import (
 	"context"
 	"fmt"
+	"time"
 
 	"github.com/cobaltcore-dev/cortex/api/v1alpha1"
 	"github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins/compute"
@@ -34,9 +35,28 @@ type ApplyResult struct {
 	RemovedReservations []v1alpha1.Reservation
 }
 
+// SlotLimitExceededError is returned by ApplyCommitmentState when the number of new reservation
+// slots would exceed MaxSlots. It is a distinct type so callers can detect and metric it separately
+// from ordinary capacity failures.
+type SlotLimitExceededError struct {
+	NewSlots int
+	Limit    int
+}
+
+func (e *SlotLimitExceededError) Error() string {
+	return fmt.Sprintf("commitment would create %d new reservation slots, exceeds limit of %d", e.NewSlots, e.Limit)
+}
+
 // ReservationManager handles CRUD operations for Reservation CRDs.
 type ReservationManager struct {
 	client.Client
+	// SlotCreationDelay adds a pause between consecutive Reservation CRD creates to spread
+	// scheduler load across time rather than bursting all creates at once.
+	SlotCreationDelay time.Duration
+	// MaxSlots caps the total number of Reservation CRDs for a single commitment.
+	// When non-zero, ApplyCommitmentState returns an error if the desired slot count would
+	// exceed this limit. Only set by the caller on the AllowRejection=true (API) path.
+	MaxSlots int
 }
 
 func NewReservationManager(k8sClient client.Client) *ReservationManager {
@@ -176,6 +196,17 @@ func (m *ReservationManager) ApplyCommitmentState(
 	}
 
 	// Phase 5 (CREATE): Create new reservations (capacity increased)
+	if deltaMemoryBytes > 0 {
+		newSlots := countNewSlots(deltaMemoryBytes, flavorGroup)
+		if m.MaxSlots > 0 && newSlots > m.MaxSlots {
+			return nil, &SlotLimitExceededError{NewSlots: newSlots, Limit: m.MaxSlots}
+		}
+		log.Info("creating reservation slots",
+			"commitmentUUID", desiredState.CommitmentUUID,
+			"slots", newSlots,
+			"slotCreationDelay", m.SlotCreationDelay,
+		)
+	}
 	for deltaMemoryBytes > 0 {
 		// Select the largest flavor that fits the remaining delta (flavors sorted descending by memory).
 		reservation := m.newReservation(desiredState, nextSlotIndex, deltaMemoryBytes, flavorGroup, creator)
@@ -196,6 +227,18 @@ func (m *ReservationManager) ApplyCommitmentState(
 		}
 
 		nextSlotIndex++
+
+		// Throttle: pause between consecutive creates to spread scheduler load.
+		// Skip after the last slot (deltaMemoryBytes <= 0) — no follow-up create to defer.
+		if m.SlotCreationDelay > 0 && deltaMemoryBytes > 0 {
+			timer := time.NewTimer(m.SlotCreationDelay)
+			select {
+			case <-ctx.Done():
+				timer.Stop()
+				return result, ctx.Err()
+			case <-timer.C:
+			}
+		}
 	}
 
 	// Phase 6 (UPDATE): Sync metadata for remaining reservations
@@ -274,6 +317,35 @@ func (m *ReservationManager) syncReservationMetadata(
 	}
 }
 
+// selectFlavor picks the largest flavor whose memory fits within deltaMemoryBytes.
+// Returns the selected flavor and its memory in bytes. If no flavor fits, returns the
+// smallest flavor with memoryBytes = deltaMemoryBytes (consumes the full remainder).
+func selectFlavor(deltaMemoryBytes int64, flavorGroup compute.FlavorGroupFeature) (flavor compute.FlavorInGroup, memoryBytes int64) {
+	flavor = flavorGroup.Flavors[len(flavorGroup.Flavors)-1]
+	memoryBytes = deltaMemoryBytes
+	for _, f := range flavorGroup.Flavors {
+		flavorBytes := int64(f.MemoryMB) * 1024 * 1024 //nolint:gosec // flavor memory from specs, realistically bounded
+		if flavorBytes <= deltaMemoryBytes {
+			flavor = f
+			memoryBytes = flavorBytes
+			break
+		}
+	}
+	return
+}
+
+// countNewSlots returns how many Reservation slots would be created to cover deltaMemoryBytes.
+// Used to pre-check MaxSlots before creating any slots, so a limit violation never leaves partial state.
+func countNewSlots(deltaMemoryBytes int64, flavorGroup compute.FlavorGroupFeature) int {
+	count := 0
+	for deltaMemoryBytes > 0 {
+		_, memoryBytes := selectFlavor(deltaMemoryBytes, flavorGroup)
+		deltaMemoryBytes -= memoryBytes
+		count++
+	}
+	return count
+}
+
 func (m *ReservationManager) newReservation(
 	state *CommitmentState,
 	slotIndex int,
@@ -290,20 +362,9 @@ func (m *ReservationManager) newReservation(
 
 	// Select largest flavor that fits remaining memory (flavors sorted descending by memory then vCPUs).
 	// This works for both fixed and varying CPU:RAM ratio groups.
-	flavorInGroup := flavorGroup.Flavors[len(flavorGroup.Flavors)-1] // default to smallest
-	memoryBytes := deltaMemoryBytes
+	flavorInGroup, memoryBytes := selectFlavor(deltaMemoryBytes, flavorGroup)
 	cpus := int64(flavorInGroup.VCPUs) //nolint:gosec // VCPUs from flavor specs, realistically bounded
 
-	for _, flavor := range flavorGroup.Flavors {
-		flavorMemoryBytes := int64(flavor.MemoryMB) * 1024 * 1024 //nolint:gosec // flavor memory from specs, realistically bounded
-		if flavorMemoryBytes <= deltaMemoryBytes {
-			flavorInGroup = flavor
-			memoryBytes = flavorMemoryBytes
-			cpus = int64(flavorInGroup.VCPUs) //nolint:gosec // VCPUs from flavor specs, realistically bounded
-			break
-		}
-	}
-
 	spec := v1alpha1.ReservationSpec{
 		Type:             v1alpha1.ReservationTypeCommittedResource,
 		SchedulingDomain: v1alpha1.SchedulingDomainNova,