From 7dbe6ec9c97459fb268973ae308d802b80a1c278 Mon Sep 17 00:00:00 2001 From: mblos Date: Fri, 22 May 2026 13:59:11 +0200 Subject: [PATCH 1/4] feat: CR safeguards, throttle CRD creation, adding limit --- cmd/manager/main.go | 1 - helm/bundles/cortex-nova/values.yaml | 4 + .../committed_resource_controller.go | 7 +- .../reservations/commitments/config.go | 28 +++---- .../commitments/reservation_manager.go | 75 ++++++++++++++++--- 5 files changed, 84 insertions(+), 31 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 28e4cf93d..63c977b5b 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -581,7 +581,6 @@ func main() { } crControllerConf := commitmentsConfig.CommittedResourceController - crControllerConf.ApplyDefaults() if err := (&commitments.CommittedResourceController{ Client: multiclusterClient, Scheme: mgr.GetScheme(), diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 3ffd30f0f..b5f20f2d4 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/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: "20ms" + # Max Reservation CRDs per CommittedResource on the API path; 0 disables the limit + maxSlotsPerCommitment: 200 committedResourceAPI: # Timeout for watching CommittedResource CRDs before rolling back watchTimeout: "15s" diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller.go b/internal/scheduling/reservations/commitments/committed_resource_controller.go index 657a9b22d..001a84236 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller.go +++ b/internal/scheduling/reservations/commitments/committed_resource_controller.go @@ -313,7 +313,12 @@ 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 { return nil, err } diff --git a/internal/scheduling/reservations/commitments/config.go b/internal/scheduling/reservations/commitments/config.go index c53ffd55e..8e18c2f9f 100644 --- a/internal/scheduling/reservations/commitments/config.go +++ b/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. diff --git a/internal/scheduling/reservations/commitments/reservation_manager.go b/internal/scheduling/reservations/commitments/reservation_manager.go index 23316fdb8..b8b47b8c3 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager.go +++ b/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" @@ -37,6 +38,13 @@ type ApplyResult struct { // 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 +184,19 @@ 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, fmt.Errorf( + "commitment would create %d new reservation slots, exceeds limit of %d", + newSlots, 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 +217,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 +307,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 +352,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, From 2d09cafad59e90d97fde104161f3c3a96cb6ea14 Mon Sep 17 00:00:00 2001 From: mblos Date: Fri, 22 May 2026 14:02:51 +0200 Subject: [PATCH 2/4] tests --- .../commitments/reservation_manager_test.go | 95 +++++++++++++++++++ 1 file changed, 95 insertions(+) diff --git a/internal/scheduling/reservations/commitments/reservation_manager_test.go b/internal/scheduling/reservations/commitments/reservation_manager_test.go index e2ce2f26c..d890bd496 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager_test.go +++ b/internal/scheduling/reservations/commitments/reservation_manager_test.go @@ -76,6 +76,7 @@ func TestApplyCommitmentState(t *testing.T) { desiredAZ string desiredDomainID string flavorGroupOverride map[string]compute.FlavorGroupFeature // nil = testFlavorGroups() + maxSlots int // 0 = no limit wantError bool wantRemovedCount int // exact count; -1 = at least one validateRemoved func(t *testing.T, removed []v1alpha1.Reservation) @@ -317,6 +318,45 @@ func TestApplyCommitmentState(t *testing.T) { } }, }, + // ---------------------------------------------------------------- + // MaxSlots limit + // ---------------------------------------------------------------- + { + name: "max slots: rejects when new slots exceed limit", + desiredMemoryGiB: 56, // 32+16+8 = 3 slots with testFlavorGroup + maxSlots: 2, + wantError: true, + }, + { + name: "max slots: allows when new slots are within limit", + desiredMemoryGiB: 56, // 32+16+8 = 3 slots + maxSlots: 3, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 3 { + t.Errorf("expected 3 slots created, got %d", len(touched)) + } + }, + }, + { + name: "max slots: only new slots counted, existing do not contribute", + existingSlots: []v1alpha1.Reservation{ + newTestCRSlot("commitment-abc123-0", 8, "", "test-group", nil), + newTestCRSlot("commitment-abc123-1", 8, "", "test-group", nil), + }, + desiredMemoryGiB: 56, // existing=16GiB, delta=40GiB → 32+8 = 2 new slots; maxSlots=2 allows it + maxSlots: 2, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + created := 0 + for _, r := range touched { + if r.Name == "commitment-abc123-2" || r.Name == "commitment-abc123-3" { + created++ + } + } + if created != 2 { + t.Errorf("expected 2 new slots created, got %d touched: %v", created, touched) + } + }, + }, } scheme := newCRTestScheme(t) @@ -329,6 +369,7 @@ func TestApplyCommitmentState(t *testing.T) { } k8sClient := newCRTestClient(scheme, objects...) manager := NewReservationManager(k8sClient) + manager.MaxSlots = tt.maxSlots flavorGroups := testFlavorGroups() if tt.flavorGroupOverride != nil { @@ -534,3 +575,57 @@ func TestNewReservation_VariableRatioGroup_SelectsLargestByMemory(t *testing.T) }) } } + +// ============================================================================ +// Tests: selectFlavor +// ============================================================================ + +func TestSelectFlavor(t *testing.T) { + fg := testFlavorGroup() // small=8GiB/4c, medium=16GiB/8c, large=32GiB/16c + + tests := []struct { + name string + deltaGiB int64 + wantFlavor string + wantMemoryGiB int64 + }{ + { + name: "exact fit: picks that flavor", + deltaGiB: 8, + wantFlavor: "small", + wantMemoryGiB: 8, + }, + { + name: "delta between small and medium: picks small", + deltaGiB: 12, + wantFlavor: "small", + wantMemoryGiB: 8, + }, + { + name: "delta larger than all flavors: picks largest, memory = largest flavor size", + deltaGiB: 100, + wantFlavor: "large", + wantMemoryGiB: 32, + }, + { + name: "delta smaller than smallest flavor: falls back, memory = full delta", + deltaGiB: 3, + wantFlavor: "small", // smallest flavor returned as fallback + wantMemoryGiB: 3, // but memory = full delta (remainder consumed) + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + deltaBytes := tt.deltaGiB * 1024 * 1024 * 1024 + flavor, memoryBytes := selectFlavor(deltaBytes, fg) + if flavor.Name != tt.wantFlavor { + t.Errorf("flavor: want %s, got %s", tt.wantFlavor, flavor.Name) + } + wantBytes := tt.wantMemoryGiB * 1024 * 1024 * 1024 + if memoryBytes != wantBytes { + t.Errorf("memoryBytes: want %d, got %d", wantBytes, memoryBytes) + } + }) + } +} From a3290c15dd4c254a6dc6621d5272b99f2c02ec81 Mon Sep 17 00:00:00 2001 From: mblos Date: Fri, 22 May 2026 14:16:32 +0200 Subject: [PATCH 3/4] metric added --- cmd/manager/main.go | 14 ++++++++------ .../commitments/committed_resource_controller.go | 10 ++++++++-- .../committed_resource_controller_monitor.go | 16 ++++++++++++++-- .../commitments/reservation_manager.go | 16 +++++++++++++--- 4 files changed, 43 insertions(+), 13 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 63c977b5b..5630964ad 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -581,18 +581,20 @@ func main() { } crControllerConf := commitmentsConfig.CommittedResourceController + + 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 { diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller.go b/internal/scheduling/reservations/commitments/committed_resource_controller.go index 001a84236..235cf6cdc 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller.go +++ b/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) { @@ -320,6 +322,10 @@ func (r *CommittedResourceController) applyReservationState(ctx context.Context, } 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) diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller_monitor.go b/internal/scheduling/reservations/commitments/committed_resource_controller_monitor.go index 116764f7f..57c6dbd77 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller_monitor.go +++ b/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) } diff --git a/internal/scheduling/reservations/commitments/reservation_manager.go b/internal/scheduling/reservations/commitments/reservation_manager.go index b8b47b8c3..01f3b2f74 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager.go +++ b/internal/scheduling/reservations/commitments/reservation_manager.go @@ -35,6 +35,18 @@ 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 @@ -187,9 +199,7 @@ func (m *ReservationManager) ApplyCommitmentState( if deltaMemoryBytes > 0 { newSlots := countNewSlots(deltaMemoryBytes, flavorGroup) if m.MaxSlots > 0 && newSlots > m.MaxSlots { - return nil, fmt.Errorf( - "commitment would create %d new reservation slots, exceeds limit of %d", - newSlots, m.MaxSlots) + return nil, &SlotLimitExceededError{NewSlots: newSlots, Limit: m.MaxSlots} } log.Info("creating reservation slots", "commitmentUUID", desiredState.CommitmentUUID, From c8692ac180f7dbac7f180f3702f761291d3f9457 Mon Sep 17 00:00:00 2001 From: mblos Date: Fri, 22 May 2026 16:32:30 +0200 Subject: [PATCH 4/4] no limit on default --- helm/bundles/cortex-nova/values.yaml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index b5f20f2d4..4a194ae50 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -174,9 +174,9 @@ cortex-scheduling-controllers: # 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: "20ms" + slotCreationDelay: "0ms" # Max Reservation CRDs per CommittedResource on the API path; 0 disables the limit - maxSlotsPerCommitment: 200 + maxSlotsPerCommitment: 0 committedResourceAPI: # Timeout for watching CommittedResource CRDs before rolling back watchTimeout: "15s"