diff --git a/cmd/manager/main.go b/cmd/manager/main.go index fca0c0550..c0b7355b7 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -626,10 +626,14 @@ func main() { monitor := reservations.NewMonitor(multiclusterClient) metrics.Registry.MustRegister(&monitor) + reservationControllerMonitor := commitments.NewReservationControllerMonitor() + metrics.Registry.MustRegister(&reservationControllerMonitor) + if err := (&commitments.CommitmentReservationController{ - Client: multiclusterClient, - Scheme: mgr.GetScheme(), - Conf: commitmentsConfig.ReservationController, + Client: multiclusterClient, + Scheme: mgr.GetScheme(), + Conf: commitmentsConfig.ReservationController, + Monitor: &reservationControllerMonitor, }).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "CommitmentReservation") os.Exit(1) diff --git a/helm/bundles/cortex-nova/templates/alerts.yaml b/helm/bundles/cortex-nova/templates/alerts.yaml index 133d8468a..169a32e7e 100644 --- a/helm/bundles/cortex-nova/templates/alerts.yaml +++ b/helm/bundles/cortex-nova/templates/alerts.yaml @@ -749,4 +749,41 @@ spec: resource router is mapping the same object to multiple clusters, or an object was created out-of-band on the wrong cluster. Investigate the affected resources and the routing configuration. + + {{- if .Values.kvm.enabled }} + - alert: CortexNovaHostReservationsOversubscribed + # Fires when the sum of running VM allocations + reservation blocks (committed + + # failover) exceeds the host's effective capacity for CPU or memory. + # This can happen due to: concurrent slot creation with stale informer cache, + # operator-driven VM migrations where the slot stays on the old host, or + # capacity changes (e.g. hardware replacement changing EffectiveCapacity). + # The 10m hold-off tolerates the known migration window: after a VM departs, + # the slot remains on the old host until the usage reconciler cleans it up. + # Note: `reserved` only counts Ready reservations — violations during the + # initial unready window (slot just created) are not captured by this alert. + expr: | + ( + max by (compute_host, availability_zone, resource) (cortex_kvm_host_capacity_usage{type="utilized"}) + + max by (compute_host, availability_zone, resource) (cortex_kvm_host_capacity_usage{type="reserved"}) + + max by (compute_host, availability_zone, resource) (cortex_kvm_host_capacity_usage{type="failover"}) + - max by (compute_host, availability_zone, resource) (cortex_kvm_host_capacity_total) + ) > 0 + for: 10m + labels: + context: committed-resource-capacity + dashboard: cortex-status-dashboard/cortex-status-dashboard + service: cortex + severity: warning + support_group: workload-management + playbook: docs/support/playbook/cortex/alerts/committed-resource-capacity + annotations: + summary: "Host {{ "{{" }} $labels.compute_host {{ "}}" }} reservation blocks exceed capacity for {{ "{{" }} $labels.resource {{ "}}" }}" + description: > + The total of running VM allocations and reservation blocks (committed resource + + failover) on host {{ "{{" }} $labels.compute_host {{ "}}" }} exceeds its effective + capacity for {{ "{{" }} $labels.resource {{ "}}" }} by {{ "{{" }} $value | humanize {{ "}}" }}. + This means the host is over-subscribed and committed resource guarantees may not be + honourable. Common causes: operator-driven VM migration with slot not yet reclaimed, a hardware + capacity change, or out of sync issues. If problem remains, inspect the reservations on this host and check the CR controller logs. + {{- end }} {{- end }} diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 6524c095e..a8a5cdc60 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -185,6 +185,15 @@ cortex-scheduling-controllers: # How long after a VM is allocated to a reservation before it is expected to appear # on the target host; allocations not confirmed within this window are removed allocationGracePeriod: "15m" + # How long to wait after first detecting host over-subscription before evicting + # reservation slots. Gives other controllers (e.g. failover) time to self-heal. + oversubscriptionGracePeriod: "3m" + # Minimum time between consecutive over-subscription checks for the same host. + # Should be shorter than oversubscriptionGracePeriod. + oversubscriptionMinCheckInterval: "1m" + # Enable host over-subscription detection and slot eviction. Acts as a kill switch + # for rollout control or when investigating capacity anomalies. Defaults to false. + enableOversubscriptionCheck: true # URL of the nova external scheduler API for placement decisions schedulerURL: "http://localhost:8080/scheduler/nova/external" # Keystone credentials used to resolve domain IDs to domain names for the diff --git a/internal/scheduling/lib/filter_monitor_test.go b/internal/scheduling/lib/filter_monitor_test.go index f709d88aa..68b47f045 100644 --- a/internal/scheduling/lib/filter_monitor_test.go +++ b/internal/scheduling/lib/filter_monitor_test.go @@ -33,6 +33,7 @@ func TestMonitorFilter(t *testing.T) { fm := monitorFilter(mockFilter, "test-filter", monitor) if fm == nil { t.Fatal("expected filter monitor, got nil") + return } if fm.filter == nil { t.Error("expected filter to be set") @@ -109,6 +110,7 @@ func TestFilterMonitor_Run(t *testing.T) { } if result == nil { t.Fatal("expected result, got nil") + return } if len(result.Activations) != 2 { t.Errorf("expected 2 activations, got %d", len(result.Activations)) diff --git a/internal/scheduling/lib/filter_validation_test.go b/internal/scheduling/lib/filter_validation_test.go index dc35c2f6a..5437cc0ff 100644 --- a/internal/scheduling/lib/filter_validation_test.go +++ b/internal/scheduling/lib/filter_validation_test.go @@ -20,6 +20,7 @@ func TestValidateFilter(t *testing.T) { if validator == nil { t.Fatal("expected validator but got nil") + return } if validator.Filter != filter { t.Error("expected filter to be set in validator") diff --git a/internal/scheduling/lib/filter_weigher_pipeline_step_test.go b/internal/scheduling/lib/filter_weigher_pipeline_step_test.go index 013946672..d25651196 100644 --- a/internal/scheduling/lib/filter_weigher_pipeline_step_test.go +++ b/internal/scheduling/lib/filter_weigher_pipeline_step_test.go @@ -121,6 +121,7 @@ func TestBaseFilterWeigherPipelineStep_IncludeAllHostsFromRequest(t *testing.T) if result == nil { t.Fatal("expected result but got nil") + return } if len(result.Activations) != tt.expectedCount { t.Errorf("expected %d activations, got %d", tt.expectedCount, len(result.Activations)) diff --git a/internal/scheduling/lib/weigher_monitor_test.go b/internal/scheduling/lib/weigher_monitor_test.go index 6f8f906e3..d6647655e 100644 --- a/internal/scheduling/lib/weigher_monitor_test.go +++ b/internal/scheduling/lib/weigher_monitor_test.go @@ -33,6 +33,7 @@ func TestMonitorWeigher(t *testing.T) { wm := monitorWeigher(mockWeigher, "test-weigher", monitor) if wm == nil { t.Fatal("expected weigher monitor, got nil") + return } if wm.weigher == nil { t.Error("expected weigher to be set") @@ -109,6 +110,7 @@ func TestWeigherMonitor_Run(t *testing.T) { } if result == nil { t.Fatal("expected result, got nil") + return } if len(result.Activations) != 2 { t.Errorf("expected 2 activations, got %d", len(result.Activations)) diff --git a/internal/scheduling/lib/weigher_validation_test.go b/internal/scheduling/lib/weigher_validation_test.go index 28dee7aec..b937a4436 100644 --- a/internal/scheduling/lib/weigher_validation_test.go +++ b/internal/scheduling/lib/weigher_validation_test.go @@ -21,6 +21,7 @@ func TestValidateWeigher(t *testing.T) { if validator == nil { t.Fatal("expected validator but got nil") + return } if validator.Weigher != weigher { t.Error("expected weigher to be set in validator") diff --git a/internal/scheduling/nova/filter_weigher_pipeline_controller_test.go b/internal/scheduling/nova/filter_weigher_pipeline_controller_test.go index ccd565f56..9e9de89c5 100644 --- a/internal/scheduling/nova/filter_weigher_pipeline_controller_test.go +++ b/internal/scheduling/nova/filter_weigher_pipeline_controller_test.go @@ -479,6 +479,7 @@ func TestFilterWeigherPipelineController_ProcessNewDecisionFromAPI(t *testing.T) ready := meta.FindStatusCondition(history.Status.Conditions, v1alpha1.HistoryConditionReady) if ready == nil { t.Fatalf("expected Ready condition to be set") + return } if ready.Status != metav1.ConditionTrue { t.Errorf("expected Ready condition status True, got %s", ready.Status) diff --git a/internal/scheduling/pods/plugins/filters/filter_node_available_test.go b/internal/scheduling/pods/plugins/filters/filter_node_available_test.go index 3eac7873c..9554c2f66 100644 --- a/internal/scheduling/pods/plugins/filters/filter_node_available_test.go +++ b/internal/scheduling/pods/plugins/filters/filter_node_available_test.go @@ -318,6 +318,7 @@ func TestNodeAvailableFilter_Run(t *testing.T) { if result == nil { t.Fatal("expected result to be non-nil") + return } if len(result.Activations) != len(tt.expected) { diff --git a/internal/scheduling/reservations/capacity_accounting.go b/internal/scheduling/reservations/capacity_accounting.go index ab305d5d5..c7b4568a9 100644 --- a/internal/scheduling/reservations/capacity_accounting.go +++ b/internal/scheduling/reservations/capacity_accounting.go @@ -75,6 +75,48 @@ func HostHasCapacityForReservation(allReservations []v1alpha1.Reservation, hv hv return true } +// HostFreeCapacity computes the remaining free capacity on hv after subtracting +// hv.Status.Allocation and UnusedReservationCapacity for all reservations on this host. +// Negative values indicate over-subscription for that resource. +// Returns nil when the hypervisor has no capacity data. +// Reservations not targeting this host (via Spec.TargetHost or Status.Host) are ignored. +func HostFreeCapacity(hostReservations []v1alpha1.Reservation, hv hv1.Hypervisor) map[hv1.ResourceName]resource.Quantity { + effCap := hv.Status.EffectiveCapacity + if effCap == nil { + effCap = hv.Status.Capacity + } + if effCap == nil { + return nil + } + + free := make(map[hv1.ResourceName]resource.Quantity, len(effCap)) + for rn, qty := range effCap { + free[rn] = qty.DeepCopy() + } + for rn, allocated := range hv.Status.Allocation { + if f, ok := free[rn]; ok { + f.Sub(allocated) + free[rn] = f + } + } + for i := range hostReservations { + res := &hostReservations[i] + if res.Spec.TargetHost == "" { + continue // evicted/unplaced; status.Host may lag until reconcile + } + if res.Spec.TargetHost != hv.Name && res.Status.Host != hv.Name { + continue + } + for rn, block := range UnusedReservationCapacity(res, false) { + if f, ok := free[rn]; ok { + f.Sub(block) + free[rn] = f + } + } + } + return free +} + // UnusedReservationCapacity returns the resources a Reservation should block on its host(s). // This is the single source of truth used by both the capacity controller and // filter_has_enough_capacity to ensure consistent accounting. diff --git a/internal/scheduling/reservations/capacity_accounting_test.go b/internal/scheduling/reservations/capacity_accounting_test.go index d13a1c400..d86e1091d 100644 --- a/internal/scheduling/reservations/capacity_accounting_test.go +++ b/internal/scheduling/reservations/capacity_accounting_test.go @@ -172,6 +172,190 @@ func TestUnusedReservationCapacity(t *testing.T) { } } +func TestHostFreeCapacity(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + cpu := func(n int64) resource.Quantity { return *resource.NewQuantity(n, resource.DecimalSI) } + + hvWithCap := func(name string, memGiB, cpuCores int64) hv1.Hypervisor { + return hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(memGiB), + hv1.ResourceCPU: cpu(cpuCores), + }, + }, + } + } + crSlot := func(name, host string, memGiB, cpuCores int64) v1alpha1.Reservation { + return v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: host, + Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(memGiB), + hv1.ResourceCPU: cpu(cpuCores), + }, + }, + Status: v1alpha1.ReservationStatus{Host: host}, + } + } + freeMemGiB := func(free map[hv1.ResourceName]resource.Quantity) int64 { + q := free[hv1.ResourceMemory] + return q.Value() / (1024 * 1024 * 1024) + } + freeCPU := func(free map[hv1.ResourceName]resource.Quantity) int64 { + q := free[hv1.ResourceCPU] + return q.Value() + } + + t.Run("no capacity data returns nil", func(t *testing.T) { + hv := hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: "host"}} + if got := HostFreeCapacity(nil, hv); got != nil { + t.Errorf("expected nil, got %v", got) + } + }) + + t.Run("no reservations and no allocation: free = effective capacity", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + free := HostFreeCapacity(nil, hv) + if freeMemGiB(free) != 1024 { + t.Errorf("expected 1024 GiB free, got %d", freeMemGiB(free)) + } + if freeCPU(free) != 256 { + t.Errorf("expected 256 CPU free, got %d", freeCPU(free)) + } + }) + + t.Run("allocation subtracted from capacity", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + hv.Status.Allocation = map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(512), + hv1.ResourceCPU: cpu(128), + } + free := HostFreeCapacity(nil, hv) + if freeMemGiB(free) != 512 { + t.Errorf("expected 512 GiB free, got %d", freeMemGiB(free)) + } + if freeCPU(free) != 128 { + t.Errorf("expected 128 CPU free, got %d", freeCPU(free)) + } + }) + + t.Run("reservation blocks subtracted", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + slots := []v1alpha1.Reservation{ + crSlot("slot-1", "host", 512, 128), + } + free := HostFreeCapacity(slots, hv) + if freeMemGiB(free) != 512 { + t.Errorf("expected 512 GiB free, got %d", freeMemGiB(free)) + } + if freeCPU(free) != 128 { + t.Errorf("expected 128 CPU free, got %d", freeCPU(free)) + } + }) + + t.Run("over-subscribed: negative free values", func(t *testing.T) { + // 5 x 1TiB slots on a 4TiB host — the production scenario + hv := hvWithCap("host", 4096, 256) + slots := []v1alpha1.Reservation{ + crSlot("slot-0", "host", 1024, 128), + crSlot("slot-1", "host", 1024, 128), + crSlot("slot-2", "host", 1024, 128), + crSlot("slot-3", "host", 1024, 128), + crSlot("slot-4", "host", 1024, 128), + } + free := HostFreeCapacity(slots, hv) + if freeMemGiB(free) != -1024 { + t.Errorf("expected -1024 GiB (over-subscribed), got %d GiB", freeMemGiB(free)) + } + if freeCPU(free) != -384 { + t.Errorf("expected -384 CPU (over-subscribed), got %d", freeCPU(free)) + } + }) + + t.Run("allocation + reservations combined", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + hv.Status.Allocation = map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(256), + hv1.ResourceCPU: cpu(64), + } + slots := []v1alpha1.Reservation{ + crSlot("slot-1", "host", 512, 128), + } + free := HostFreeCapacity(slots, hv) + // 1024 - 256 (alloc) - 512 (slot) = 256 GiB free + if freeMemGiB(free) != 256 { + t.Errorf("expected 256 GiB free, got %d", freeMemGiB(free)) + } + // 256 - 64 (alloc) - 128 (slot) = 64 free + if freeCPU(free) != 64 { + t.Errorf("expected 64 CPU free, got %d", freeCPU(free)) + } + }) + + t.Run("confirmed VM reduces slot block (not double counted)", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + // 256 GiB confirmed VM already counted in Allocation + hv.Status.Allocation = map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(256), + } + // 512 GiB slot with 256 GiB confirmed VM → block = 512-256 = 256 GiB + slot := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-1"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-1": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(256)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host"}, + }, + }, + } + free := HostFreeCapacity([]v1alpha1.Reservation{slot}, hv) + // 1024 - 256 (alloc/vm) - 256 (remaining slot block) = 512 + if freeMemGiB(free) != 512 { + t.Errorf("expected 512 GiB free, got %d", freeMemGiB(free)) + } + }) + + t.Run("reservations on other hosts are ignored even if passed in", func(t *testing.T) { + hv := hvWithCap("host", 1024, 256) + slots := []v1alpha1.Reservation{ + crSlot("slot-other", "other-host", 1024, 256), // different host — must not block + } + free := HostFreeCapacity(slots, hv) + if freeMemGiB(free) != 1024 { + t.Errorf("expected 1024 GiB free (other host ignored), got %d", freeMemGiB(free)) + } + }) + + t.Run("falls back to Capacity when EffectiveCapacity nil", func(t *testing.T) { + hv := hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host"}, + Status: hv1.HypervisorStatus{ + Capacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(512), + }, + }, + } + free := HostFreeCapacity(nil, hv) + if freeMemGiB(free) != 512 { + t.Errorf("expected 512 GiB, got %d", freeMemGiB(free)) + } + }) +} + func TestHostHasCapacityForReservation(t *testing.T) { gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } cpu := func(n int64) resource.Quantity { return *resource.NewQuantity(n, resource.DecimalSI) } diff --git a/internal/scheduling/reservations/commitments/api/report_capacity_test.go b/internal/scheduling/reservations/commitments/api/report_capacity_test.go index 4bcb71056..7d65212b2 100644 --- a/internal/scheduling/reservations/commitments/api/report_capacity_test.go +++ b/internal/scheduling/reservations/commitments/api/report_capacity_test.go @@ -250,10 +250,12 @@ func TestCapacityCalculator(t *testing.T) { ramRes := report.Resources["hw_version_test-group_ram"] if ramRes == nil { t.Fatal("missing hw_version_test-group_ram") + return } az := ramRes.PerAZ[tc.checkAZ] if az == nil { t.Fatalf("missing entry for AZ %s", tc.checkAZ) + return } if az.Capacity != tc.wantCapacity { t.Errorf("capacity = %d, want %d", az.Capacity, tc.wantCapacity) diff --git a/internal/scheduling/reservations/commitments/api/usage_test.go b/internal/scheduling/reservations/commitments/api/usage_test.go index eca1f0f9a..a39ded2f9 100644 --- a/internal/scheduling/reservations/commitments/api/usage_test.go +++ b/internal/scheduling/reservations/commitments/api/usage_test.go @@ -479,6 +479,7 @@ func TestUsageMultipleCalculation_FloorDivision(t *testing.T) { ramResource := report.Resources[liquid.ResourceName("hw_version_hw_2101_ram")] if ramResource == nil { t.Fatal("hw_version_hw_2101_ram resource not found") + return } var totalRAM uint64 for _, azReport := range ramResource.PerAZ { @@ -491,6 +492,7 @@ func TestUsageMultipleCalculation_FloorDivision(t *testing.T) { coresResource := report.Resources[liquid.ResourceName("hw_version_hw_2101_cores")] if coresResource == nil { t.Fatal("hw_version_hw_2101_cores resource not found") + return } var totalCores uint64 for _, azReport := range coresResource.PerAZ { @@ -503,6 +505,7 @@ func TestUsageMultipleCalculation_FloorDivision(t *testing.T) { instancesResource := report.Resources[liquid.ResourceName("hw_version_hw_2101_instances")] if instancesResource == nil { t.Fatal("hw_version_hw_2101_instances resource not found") + return } var totalInstances uint64 for _, azReport := range instancesResource.PerAZ { diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller_test.go b/internal/scheduling/reservations/commitments/committed_resource_controller_test.go index 293f074a9..a77fd741c 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller_test.go +++ b/internal/scheduling/reservations/commitments/committed_resource_controller_test.go @@ -22,6 +22,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client/fake" "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" ) // ============================================================================ @@ -137,6 +138,24 @@ func newCRTestClient(scheme *runtime.Scheme, objects ...client.Object) client.Cl } return uuids }). + WithIndex(&v1alpha1.Reservation{}, reservations.IdxReservationByHost, func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok { + return nil + } + hosts := make(map[string]struct{}) + if res.Spec.TargetHost != "" { + hosts[res.Spec.TargetHost] = struct{}{} + } + if res.Status.Host != "" { + hosts[res.Status.Host] = struct{}{} + } + result := make([]string, 0, len(hosts)) + for h := range hosts { + result = append(result, h) + } + return result + }). Build() } @@ -1195,6 +1214,7 @@ func TestCommittedResourceController_SetAcceptedIdempotent(t *testing.T) { cond := meta.FindStatusCondition(after1.Status.Conditions, v1alpha1.CommittedResourceConditionReady) if cond == nil { t.Fatalf("Ready condition not set after first reconcile") + return } if cond.ObservedGeneration != 1 { t.Errorf("ObservedGeneration: want 1, got %d", cond.ObservedGeneration) diff --git a/internal/scheduling/reservations/commitments/config.go b/internal/scheduling/reservations/commitments/config.go index 269ba3e4a..a130ca11b 100644 --- a/internal/scheduling/reservations/commitments/config.go +++ b/internal/scheduling/reservations/commitments/config.go @@ -62,6 +62,19 @@ type ReservationControllerConfig struct { // reservation during which it's expected to appear on the target host. // VMs not confirmed within this period are considered stale and removed. AllocationGracePeriod metav1.Duration `json:"allocationGracePeriod"` + // OversubscriptionGracePeriod is how long to wait after detecting host over-subscription + // before evicting slots. Gives other controllers (e.g. failover) time to self-heal. + OversubscriptionGracePeriod metav1.Duration `json:"oversubscriptionGracePeriod,omitempty"` + // OversubscriptionMinCheckInterval is the minimum time between consecutive over-subscription + // checks for the same host. Independent of RequeueIntervalActive so it can be tuned + // to a value shorter than the grace period without affecting normal reconcile cadence. + // Defaults to 30s. + OversubscriptionMinCheckInterval metav1.Duration `json:"oversubscriptionMinCheckInterval,omitempty"` + // EnableOversubscriptionCheck enables the host over-subscription detection and slot eviction + // logic. When false, no oversubscription checks are performed and no slots are evicted. + // Acts as a kill switch for rollout control or when investigating capacity anomalies. + // Defaults to false (opt-in). + EnableOversubscriptionCheck bool `json:"enableOversubscriptionCheck,omitempty"` // SchedulerURL is the endpoint of the nova external scheduler. SchedulerURL string `json:"schedulerURL"` // PipelineDefault is the fallback pipeline when no FlavorGroupPipelines entry matches. diff --git a/internal/scheduling/reservations/commitments/field_index.go b/internal/scheduling/reservations/commitments/field_index.go index 1237d0da5..1a4ca1a3b 100644 --- a/internal/scheduling/reservations/commitments/field_index.go +++ b/internal/scheduling/reservations/commitments/field_index.go @@ -131,7 +131,6 @@ func indexProjectQuotaByProjectID(ctx context.Context, mcl *multicluster.Client) return err } -// indexReservationByAllocationVMUUID registers an index over all VM UUIDs present in // Spec.CommittedResourceReservation.Allocations. This allows the reservation controller // to efficiently find all other Reservation CRDs carrying a specific VM UUID without // scanning every reservation in the cluster. diff --git a/internal/scheduling/reservations/commitments/integration_test.go b/internal/scheduling/reservations/commitments/integration_test.go index ce968b4f4..7ab025122 100644 --- a/internal/scheduling/reservations/commitments/integration_test.go +++ b/internal/scheduling/reservations/commitments/integration_test.go @@ -20,6 +20,7 @@ package commitments import ( "context" "encoding/json" + "fmt" "net/http" "net/http/httptest" "strings" @@ -443,6 +444,24 @@ func newIntgEnv(t *testing.T, initialObjects []client.Object, schedulerFn http.H } return uuids }). + WithIndex(&v1alpha1.Reservation{}, reservations.IdxReservationByHost, func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok { + return nil + } + hosts := make(map[string]struct{}) + if res.Spec.TargetHost != "" { + hosts[res.Spec.TargetHost] = struct{}{} + } + if res.Status.Host != "" { + hosts[res.Status.Host] = struct{}{} + } + result := make([]string, 0, len(hosts)) + for h := range hosts { + result = append(result, h) + } + return result + }). Build() schedulerSrv := httptest.NewServer(schedulerFn) @@ -456,14 +475,17 @@ func newIntgEnv(t *testing.T, initialObjects []client.Object, schedulerFn http.H }, VMSource: vmSource, } + monitor := NewReservationControllerMonitor() resCtrl := &CommitmentReservationController{ Client: k8sClient, Scheme: scheme, Conf: ReservationControllerConfig{ - SchedulerURL: schedulerSrv.URL, - AllocationGracePeriod: metav1.Duration{Duration: 15 * time.Minute}, - RequeueIntervalActive: metav1.Duration{Duration: 5 * time.Minute}, + SchedulerURL: schedulerSrv.URL, + AllocationGracePeriod: metav1.Duration{Duration: 15 * time.Minute}, + RequeueIntervalActive: metav1.Duration{Duration: 5 * time.Minute}, + EnableOversubscriptionCheck: true, }, + Monitor: &monitor, } if err := resCtrl.Init(context.Background(), resCtrl.Conf); err != nil { t.Fatalf("resCtrl.Init: %v", err) @@ -491,7 +513,10 @@ func (e *intgEnv) reconcileReservation(t *testing.T, resName string) { t.Helper() req := ctrl.Request{NamespacedName: types.NamespacedName{Name: resName}} if _, err := e.resController.Reconcile(context.Background(), req); err != nil { - t.Fatalf("reservation reconcile %s: %v", resName, err) + // "no hosts found" is a retriable error surfaced deliberately for backoff — not a test failure. + if !strings.Contains(err.Error(), "no hosts found") { + t.Fatalf("reservation reconcile %s: %v", resName, err) + } } } @@ -1010,6 +1035,7 @@ func TestCRLifecycle(t *testing.T) { cond := meta.FindStatusCondition(final.Status.Conditions, v1alpha1.CommittedResourceConditionReady) if cond == nil { t.Fatalf("no Ready condition") + return } if cond.Reason == v1alpha1.CommittedResourceReasonRejected { t.Errorf("AllowRejection=false: CR must not transition to Rejected, got Reason=%s", cond.Reason) @@ -1184,6 +1210,7 @@ func TestCRLifecycle(t *testing.T) { cond := meta.FindStatusCondition(final.Status.Conditions, v1alpha1.CommittedResourceConditionReady) if cond == nil { t.Fatalf("no Ready condition after retries") + return } if cond.Reason == v1alpha1.CommittedResourceReasonRejected { t.Errorf("AllowRejection=false: CR must not be Rejected, got Reason=%s", cond.Reason) @@ -1231,3 +1258,139 @@ func TestCRScheduling_SetsReserveForCommittedResourceIntent(t *testing.T) { t.Errorf("CR slot scheduling must set _nova_check_type=%q, got %q", schedulerdelegationapi.ReserveForCommittedResourceIntent, hint) } } + +func TestCROversubscriptionRemediation(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + cpu := func(n int64) resource.Quantity { return *resource.NewQuantity(n, resource.DecimalSI) } + + hv := &hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{ + Name: "host-1", + Labels: map[string]string{"topology.kubernetes.io/zone": "az1"}, + }, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(200), + hv1.ResourceCPU: cpu(200), + }, + }, + } + + objects := []client.Object{hv} + for i := range 20 { + memGiB := int64(10 + i) + cores := int64(10 + i) + name := fmt.Sprintf("slot-%02d", i) + + slot := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(memGiB), + hv1.ResourceCPU: cpu(cores), + }, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + ResourceName: "test-flavor", + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + Conditions: []metav1.Condition{{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionTrue, + Reason: "ReservationActive", + }}, + }, + } + if i < 10 { + // slots 0-9: allocated — have a confirmed VM + vmID := fmt.Sprintf("vm-%02d", i) + slot.Spec.CommittedResourceReservation.Allocations = map[string]v1alpha1.CommittedResourceAllocation{ + vmID: {Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(memGiB), + hv1.ResourceCPU: cpu(cores), + }}, + } + slot.Status.CommittedResourceReservation = &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{vmID: "host-1"}, + } + } + objects = append(objects, slot) + } + + var schedulerCalls atomic.Int32 + schedulerFn := func(w http.ResponseWriter, r *http.Request) { + schedulerCalls.Add(1) + intgRejectScheduler(w, r) // reject so evicted slots don't land back on host-1 + } + env := newIntgEnv(t, append(objects, newTestFlavorKnowledge()), schedulerFn, nil) + defer env.close() + + // Reconcile slot-10 — triggers oversubscription detection. + env.reconcileReservation(t, "slot-10") + + if _, ok := env.resController.oversubscriptionFirstSeen["host-1"]; !ok { + t.Fatal("expected oversubscription to be detected after first reconcile") + } + + // Drive remediation: fast-forward grace period and reset rate limit per cycle. + // Rotate trigger slot each time — avoids fake client index staleness + // (status patches don't update field indexes in the fake client). + resolved := false + for i := range 25 { + triggerSlot := fmt.Sprintf("slot-%02d", 10+i%10) + env.resController.oversubscriptionFirstSeen["host-1"] = time.Now().Add(-3 * time.Minute) + env.resController.oversubscriptionLastCheckedAt["host-1"] = time.Time{} + env.reconcileReservation(t, triggerSlot) + + if _, still := env.resController.oversubscriptionFirstSeen["host-1"]; !still { + resolved = true + break + } + } + if !resolved { + t.Fatal("expected oversubscription to be resolved after remediation cycles") + } + + var resList v1alpha1.ReservationList + if err := env.k8sClient.List(context.Background(), &resList); err != nil { + t.Fatalf("list reservations: %v", err) + } + + var placedSlots []v1alpha1.Reservation + for _, r := range resList.Items { + if r.Spec.TargetHost == "host-1" || r.Status.Host == "host-1" { + placedSlots = append(placedSlots, r) + } + } + free := reservations.HostFreeCapacity(placedSlots, *hv) + zero := resource.MustParse("0") + for rn, f := range free { + if f.Cmp(zero) < 0 { + t.Errorf("host still over-subscribed for %s by %s after remediation", rn, f.String()) + } + } + + for _, r := range resList.Items { + if r.Spec.TargetHost == "" && + r.Spec.CommittedResourceReservation != nil && + len(r.Spec.CommittedResourceReservation.Allocations) > 0 { + t.Errorf("slot %s has allocations but was evicted", r.Name) + } + } + + // Two reconciles needed: first clears status (PlacementRevoked), second triggers placement. + callsBefore := schedulerCalls.Load() + for _, r := range resList.Items { + if r.Spec.TargetHost == "" { + env.reconcileReservation(t, r.Name) // revoke status + env.reconcileReservation(t, r.Name) // trigger placement + break + } + } + if schedulerCalls.Load() <= callsBefore { + t.Error("expected evicted slot to trigger a scheduler call on next reconcile") + } +} diff --git a/internal/scheduling/reservations/commitments/payg_rollback_test.go b/internal/scheduling/reservations/commitments/payg_rollback_test.go index c9962436a..f1f7efeb0 100644 --- a/internal/scheduling/reservations/commitments/payg_rollback_test.go +++ b/internal/scheduling/reservations/commitments/payg_rollback_test.go @@ -5,6 +5,7 @@ package commitments import ( "context" + "strings" "testing" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -82,10 +83,10 @@ func TestCRLifecycle_PAYGRollback(t *testing.T) { } for _, res := range resList.Items { resReq := ctrl.Request{NamespacedName: types.NamespacedName{Name: res.Name}} - if _, err := env.resController.Reconcile(ctx, resReq); err != nil { + if _, err := env.resController.Reconcile(ctx, resReq); err != nil && !strings.Contains(err.Error(), "no hosts found") { t.Fatalf("reservation reconcile %s (pass 1): %v", res.Name, err) } - if _, err := env.resController.Reconcile(ctx, resReq); err != nil { + if _, err := env.resController.Reconcile(ctx, resReq); err != nil && !strings.Contains(err.Error(), "no hosts found") { t.Fatalf("reservation reconcile %s (pass 2): %v", res.Name, err) } } diff --git a/internal/scheduling/reservations/commitments/reservation_controller.go b/internal/scheduling/reservations/commitments/reservation_controller.go index f6bf5a385..07f14b4d4 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller.go +++ b/internal/scheduling/reservations/commitments/reservation_controller.go @@ -6,9 +6,13 @@ package commitments import ( "context" "fmt" + "reflect" + "sort" + "sync" "time" "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -22,6 +26,8 @@ import ( "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" + "net/http" + schedulerdelegationapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/scheduling" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -32,7 +38,6 @@ import ( hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "github.com/go-logr/logr" "github.com/gophercloud/gophercloud/v2" - "net/http" ) // CommitmentReservationController reconciles commitment Reservation objects @@ -49,6 +54,15 @@ type CommitmentReservationController struct { // domain_name scheduler hint can be populated for filter_external_customer. // Nil when KeystoneSecretRef is not configured; hint is omitted in that case. DomainResolver DomainResolver + // Monitor reports over-subscription violations as Prometheus metrics. + // Nil disables over-subscription detection entirely. + Monitor *ReservationControllerMonitor + + // oversubscription tracking — mu protects the three maps below + oversubscriptionMu sync.Mutex + oversubscriptionLastCheckedAt map[string]time.Time + oversubscriptionPendingCheck map[string]bool + oversubscriptionFirstSeen map[string]time.Time } // echoParentGeneration copies Spec.CommittedResourceReservation.ParentGeneration to @@ -114,6 +128,28 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr } if res.IsReady() { + // Spec.TargetHost was cleared (e.g. oversubscription eviction) — revoke status so the + // slot re-enters the placement flow on the next reconcile. + if res.Spec.TargetHost == "" { + old := res.DeepCopy() + res.Status.Host = "" + if res.Status.CommittedResourceReservation != nil { + res.Status.CommittedResourceReservation.Allocations = nil + } + meta.SetStatusCondition(&res.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionFalse, + Reason: "PlacementRevoked", + Message: "target host was cleared; pending re-placement", + }) + if err := r.Status().Patch(ctx, &res, client.MergeFrom(old)); client.IgnoreNotFound(err) != nil { + return ctrl.Result{}, err + } + logger.Info("revoked ready status after placement eviction, slot re-enters placement flow", + "component", "oversubscription-check") + return ctrl.Result{}, nil + } + logger.V(1).Info("reservation is active, verifying allocations") // Sync ObservedParentGeneration if the CR controller bumped ParentGeneration since @@ -141,6 +177,10 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr if result.HasAllocationsInGracePeriod { return ctrl.Result{RequeueAfter: r.Conf.RequeueIntervalGracePeriod.Duration}, nil } + // Check over-subscription after allocation verification (HV watch path). + if requeueAfter := r.runOversubscriptionCheck(ctx, res.Status.Host); requeueAfter > 0 { + return ctrl.Result{RequeueAfter: requeueAfter}, nil + } return ctrl.Result{RequeueAfter: r.Conf.RequeueIntervalActive.Duration}, nil } @@ -205,6 +245,10 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr return ctrl.Result{}, nil } logger.Info("synced spec to status and marked ready", "host", res.Status.Host) + // Check over-subscription now that this slot is placed and Ready. + if requeueAfter := r.runOversubscriptionCheck(ctx, res.Status.Host); requeueAfter > 0 { + return ctrl.Result{RequeueAfter: requeueAfter}, nil + } // Return and let next reconcile handle allocation verification return ctrl.Result{}, nil } @@ -295,7 +339,14 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr logger.Info("selected pipeline for CR reservation", "flavorName", resourceName, "flavorGroup", flavorGroupName, - "pipeline", pipelineName) + "pipeline", pipelineName, + "reason", func() string { + cond := meta.FindStatusCondition(res.Status.Conditions, v1alpha1.ReservationConditionReady) + if cond != nil { + return cond.Reason + } + return "initial" + }()) // Use the SchedulerClient to schedule the reservation scheduleReq := reservations.ScheduleReservationRequest{ @@ -328,7 +379,7 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr } if len(scheduleResp.Hosts) == 0 { - logger.Info("no hosts found for reservation", "reservation", res.Name, "flavorName", resourceName) + logger.Info("no hosts found for reservation, will retry", "reservation", res.Name, "flavorName", resourceName) old := res.DeepCopy() meta.SetStatusCondition(&res.Status.Conditions, metav1.Condition{ Type: v1alpha1.ReservationConditionReady, @@ -347,14 +398,14 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr // Object was deleted, no need to continue return ctrl.Result{}, nil } - return ctrl.Result{}, nil // No need to requeue, we didn't find a host. + return ctrl.Result{}, fmt.Errorf("no hosts found for reservation %s (flavor %s)", res.Name, resourceName) } // Update the reservation Spec with the found host (idx 0) // Only update Spec here - the Status will be synced in the next reconcile cycle // This avoids race conditions from doing two patches in one reconcile host := scheduleResp.Hosts[0] - logger.Info("found host for reservation", "host", host) + logger.Info("found host for reservation", "host", host, "flavorName", resourceName) old := res.DeepCopy() res.Spec.TargetHost = host @@ -718,7 +769,7 @@ func (r *CommitmentReservationController) getPipelineForFlavorGroup(flavorGroupN func (r *CommitmentReservationController) hypervisorToReservations(ctx context.Context, obj client.Object) []reconcile.Request { hvName := obj.GetName() var reservationList v1alpha1.ReservationList - if err := r.List(ctx, &reservationList); err != nil { + if err := r.List(ctx, &reservationList, client.MatchingFields{reservations.IdxReservationByHost: hvName}); err != nil { logf.FromContext(ctx).Error(err, "failed to list reservations for hypervisor", "hypervisor", hvName) return nil } @@ -811,6 +862,28 @@ var commitmentReservationPredicate = predicate.Funcs{ }, } +// hvCapacityChangePredicate fires when Status.Instances, Status.Allocation, +// Status.EffectiveCapacity, or Status.Capacity changes on a Hypervisor. Instances covers +// VM presence (used by allocation verification); Allocation, EffectiveCapacity, and Capacity +// cover capacity accounting (used by the over-subscription check; Capacity is the fallback +// when EffectiveCapacity is nil). +var hvCapacityChangePredicate = predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { _, ok := e.Object.(*hv1.Hypervisor); return ok }, + DeleteFunc: func(e event.DeleteEvent) bool { _, ok := e.Object.(*hv1.Hypervisor); return ok }, + GenericFunc: func(e event.GenericEvent) bool { _, ok := e.Object.(*hv1.Hypervisor); return ok }, + UpdateFunc: func(e event.UpdateEvent) bool { + oldHV, ok1 := e.ObjectOld.(*hv1.Hypervisor) + newHV, ok2 := e.ObjectNew.(*hv1.Hypervisor) + if !ok1 || !ok2 { + return false + } + return !reflect.DeepEqual(oldHV.Status.Instances, newHV.Status.Instances) || + !reflect.DeepEqual(oldHV.Status.Allocation, newHV.Status.Allocation) || + !reflect.DeepEqual(oldHV.Status.EffectiveCapacity, newHV.Status.EffectiveCapacity) || + !reflect.DeepEqual(oldHV.Status.Capacity, newHV.Status.Capacity) + }, +} + // SetupWithManager sets up the controller with the Manager. func (r *CommitmentReservationController) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client) error { if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { @@ -825,6 +898,9 @@ func (r *CommitmentReservationController) SetupWithManager(mgr ctrl.Manager, mcl if err := indexReservationByAllocationVMUUID(context.Background(), mcl); err != nil { return fmt.Errorf("failed to set up reservation allocation VM UUID index: %w", err) } + if err := reservations.IndexReservationByHost(context.Background(), mcl); err != nil { + return fmt.Errorf("failed to set up reservation by host index: %w", err) + } // Use WatchesMulticluster to watch Reservations across all configured clusters // (home + remotes). This is required because Reservation CRDs may be stored @@ -848,6 +924,7 @@ func (r *CommitmentReservationController) SetupWithManager(mgr ctrl.Manager, mcl bldr, err = bldr.WatchesMulticluster( &hv1.Hypervisor{}, handler.EnqueueRequestsFromMapFunc(r.hypervisorToReservations), + hvCapacityChangePredicate, ) if err != nil { return err @@ -863,3 +940,336 @@ func (r *CommitmentReservationController) SetupWithManager(mgr ctrl.Manager, mcl }). Complete(r) } + +// runOversubscriptionCheck detects host over-subscription and drives remediation. +// Rate-limited per host: skipped checks mark the host pending so the next reconcile retries. +// Returns non-zero when the caller should requeue (grace period pending or after eviction). +func (r *CommitmentReservationController) runOversubscriptionCheck(ctx context.Context, host string) time.Duration { + if host == "" || r.Monitor == nil || !r.Conf.EnableOversubscriptionCheck { + return 0 + } + + r.oversubscriptionMu.Lock() + defer r.oversubscriptionMu.Unlock() + + logger := LoggerFromContext(ctx).WithValues("component", "oversubscription-check", "host", host) + + gracePeriod := r.Conf.OversubscriptionGracePeriod.Duration + if gracePeriod == 0 { + gracePeriod = 2 * time.Minute + } + minCheckInterval := r.Conf.OversubscriptionMinCheckInterval.Duration + if minCheckInterval == 0 { + minCheckInterval = 30 * time.Second + } + + if r.oversubscriptionLastCheckedAt == nil { + r.oversubscriptionLastCheckedAt = make(map[string]time.Time) + r.oversubscriptionPendingCheck = make(map[string]bool) + r.oversubscriptionFirstSeen = make(map[string]time.Time) + } + + // Rate limit: if checked recently and if pending flag marks already dirty + if timeSinceLastCheck := time.Since(r.oversubscriptionLastCheckedAt[host]); timeSinceLastCheck < minCheckInterval { + if !r.oversubscriptionPendingCheck[host] { + r.oversubscriptionPendingCheck[host] = true + requeue := minCheckInterval - timeSinceLastCheck + time.Second + logger.V(1).Info("over-subscription check rate-limited, requeue scheduled", + "lastCheckedAgo", timeSinceLastCheck.Round(time.Second), + "requeueIn", requeue.Round(time.Second)) + return requeue + } + // already dirty, so someone else requeued already + return 0 + } + + var hv hv1.Hypervisor + if err := r.Get(ctx, client.ObjectKey{Name: host}, &hv); err != nil { + logger.Error(err, "failed to get hypervisor for over-subscription check") + return 0 + } + var hostReservations v1alpha1.ReservationList + if err := r.List(ctx, &hostReservations, client.MatchingFields{reservations.IdxReservationByHost: host}); err != nil { + logger.Error(err, "failed to list reservations for over-subscription check") + return 0 + } + + logger.Info("running over-subscription check", + "slotCount", len(hostReservations.Items), + "lastCheckedAgo", time.Since(r.oversubscriptionLastCheckedAt[host]).Round(time.Second), + "violationFirstSeen", func() string { + if t := r.oversubscriptionFirstSeen[host]; !t.IsZero() { + return time.Since(t).Round(time.Second).String() + } + return "none" + }()) + + // Mark check as done and clear dirty flag before delegating. + r.oversubscriptionLastCheckedAt[host] = time.Now() + r.oversubscriptionPendingCheck[host] = false + firstSeen := r.oversubscriptionFirstSeen[host] + + evicted, resolved, err := r.checkHostOversubscription(ctx, host, hostReservations.Items, hv, r.Monitor, gracePeriod, firstSeen) + if err != nil { + logger.Error(err, "over-subscription check failed") + return 0 + } + + // No violation — clear grace period state. + if resolved { + if !firstSeen.IsZero() { + logger.Info("host over-subscription resolved", + "violationDuration", time.Since(firstSeen).Round(time.Second)) + } else { + logger.Info("host over-subscription check: no violations detected") + } + delete(r.oversubscriptionFirstSeen, host) + return 0 + } + + // Slot evicted — reset grace period so next eviction waits a full interval. + if evicted { + r.oversubscriptionFirstSeen[host] = time.Now() + logger.Info("slot evicted, next remediation check in", "nextCheckIn", gracePeriod) + return gracePeriod + } + + // Violation detected for the first time — start grace period, requeue after it. + if firstSeen.IsZero() { + r.oversubscriptionFirstSeen[host] = time.Now() + logger.Info("violation detected first time, starting grace period", "gracePeriod", gracePeriod) + return gracePeriod + } + + // Grace period still running — requeue with remaining time. + if elapsed := time.Since(firstSeen); elapsed < gracePeriod { + var remaining = gracePeriod - elapsed + logger.Info("violation still present, grace period running", "elapsed", elapsed.Round(time.Second), "remaining", remaining.Round(time.Second)) + return remaining + } + + // Grace period elapsed but checkHostOversubscription did not evict (no candidates). + return 0 +} + +// computeViolations returns the per-resource excess (positive quantity) for any resource where +// the host is over-subscribed. Returns nil if capacity data is unavailable, empty map if no violation. +func computeViolations(allReservations []v1alpha1.Reservation, hv hv1.Hypervisor) map[hv1.ResourceName]resource.Quantity { + free := reservations.HostFreeCapacity(allReservations, hv) + if free == nil { + return nil + } + zero := resource.MustParse("0") + violations := make(map[hv1.ResourceName]resource.Quantity) + for rn, f := range free { + if f.Cmp(zero) < 0 { + excess := f.DeepCopy() + excess.Neg() + violations[rn] = excess + } + } + return violations +} + +// selectEvictionTarget picks the best slot to evict from allReservations given the active violations. +func (r *CommitmentReservationController) selectEvictionTarget( + ctx context.Context, + allReservations []v1alpha1.Reservation, + violations map[hv1.ResourceName]resource.Quantity, +) *v1alpha1.Reservation { + + logger := LoggerFromContext(ctx).WithValues("component", "oversubscription-check") + var selectedRes *v1alpha1.Reservation + + var unallocated, allocated []*v1alpha1.Reservation + for i := range allReservations { + res := &allReservations[i] + if res.Spec.Type != v1alpha1.ReservationTypeCommittedResource { + continue + } + if res.Spec.TargetHost == "" { + continue // already evicted, pending status cleanup + } + if res.Spec.CommittedResourceReservation == nil { + continue // no CR payload — nothing to evict + } + if len(res.Spec.CommittedResourceReservation.Allocations) == 0 { + unallocated = append(unallocated, res) + } else { + allocated = append(allocated, res) + } + } + + // Sort unallocated by memory asc, then CPU asc — smallest slots are easiest to re-place. + sort.SliceStable(unallocated, func(i, j int) bool { + ri, rj := unallocated[i].Spec.Resources, unallocated[j].Spec.Resources + mi, mj := ri[hv1.ResourceMemory], rj[hv1.ResourceMemory] + if c := mi.Cmp(mj); c != 0 { + return c < 0 + } + ci, cj := ri[hv1.ResourceCPU], rj[hv1.ResourceCPU] + return ci.Cmp(cj) < 0 + }) + + // First choice: smallest unallocated slot (easiest to re-place). + if len(unallocated) > 0 { + selectedRes = unallocated[0] + } + + // Last resort: allocated slot with the largest fraction of unused capacity. + if selectedRes == nil && len(allocated) > 0 { + unusedCapRatioBuckets := make([]map[hv1.ResourceName]int64, len(allocated)) + for i, res := range allocated { + unusedCap := reservations.UnusedReservationCapacity(res, false) + buckets := make(map[hv1.ResourceName]int64) + for rn, total := range res.Spec.Resources { + if t := total.Value(); t != 0 { + unused := unusedCap[rn] + buckets[rn] = unused.Value() * 10 / t + } + } + unusedCapRatioBuckets[i] = buckets + } + // Sort: 1st by unused mem ratio desc, 2nd by unused CPU ratio desc, 3rd by total mem asc. + sort.SliceStable(allocated, func(i, j int) bool { + if d := unusedCapRatioBuckets[i][hv1.ResourceMemory] - unusedCapRatioBuckets[j][hv1.ResourceMemory]; d != 0 { + return d > 0 + } + if d := unusedCapRatioBuckets[i][hv1.ResourceCPU] - unusedCapRatioBuckets[j][hv1.ResourceCPU]; d != 0 { + return d > 0 + } + mi, mj := allocated[i].Spec.Resources[hv1.ResourceMemory], allocated[j].Spec.Resources[hv1.ResourceMemory] + return mi.Cmp(mj) < 0 + }) + selectedRes = allocated[0] + } + + if selectedRes == nil { + logger.Info("no eviction target found") + return nil + } + logger.Info("eviction target selected", + "reservation", selectedRes.Name, + "total unallocated slots", len(unallocated), + "total allocated slots", len(allocated), + "slot resources", selectedRes.Spec.Resources, + "violations", violations, + ) + return selectedRes +} + +// checkHostOversubscription detects host over-subscription and evicts one slot if grace period elapsed. +// firstSeen is the time the violation was first detected (zero if not yet seen). +// Returns (evicted, resolved, err): evicted=slot was unplaced, resolved=no violation. +func (r *CommitmentReservationController) checkHostOversubscription( + ctx context.Context, + host string, + allReservations []v1alpha1.Reservation, + hv hv1.Hypervisor, + monitor *ReservationControllerMonitor, + gracePeriod time.Duration, + firstSeen time.Time, +) (evicted, resolved bool, err error) { + + logger := LoggerFromContext(ctx).WithValues("component", "oversubscription-check", "host", host) + az := hv.Labels["topology.kubernetes.io/zone"] + + updateMonitor := func(v map[hv1.ResourceName]resource.Quantity) { + monitor.ClearHost(host, az) + for rn, excess := range v { + monitor.SetOversubscribed(host, az, string(rn), float64(excess.Value())) + } + } + + violations := computeViolations(allReservations, hv) + if violations == nil { + monitor.ClearHost(host, az) + return false, false, nil + } + if len(violations) == 0 { + monitor.ClearHost(host, az) + return false, true, nil + } + + if firstSeen.IsZero() { + logger.Info("host over-subscribed, starting grace period", + "gracePeriod", gracePeriod, + "violations", violations) + updateMonitor(violations) + return false, false, nil + } + + elapsed := time.Since(firstSeen) + if elapsed < gracePeriod { + logger.Info("host over-subscribed, grace period in progress", + "elapsed", elapsed.Round(time.Second), + "remaining", (gracePeriod - elapsed).Round(time.Second)) + updateMonitor(violations) + return false, false, nil + } + + logger.Info("host over-subscribed, evicting one slot", "violations", violations) + + target := r.selectEvictionTarget(ctx, allReservations, violations) + if target == nil { + logger.Info("host over-subscribed but no evictable CR reservation slots found — manual intervention required", + "violations", violations) + updateMonitor(violations) + return false, false, nil + } + + freed, err := r.unplaceReservation(ctx, target) + if err != nil { + logger.Error(err, "failed to unplace reservation for over-subscription remediation", "reservation", target.Name) + updateMonitor(violations) + return false, false, err + } + + logger.Info("evicted slot for over-subscription remediation", + "reservation", target.Name, + "hasAllocations", target.Spec.CommittedResourceReservation != nil && len(target.Spec.CommittedResourceReservation.Allocations) > 0, + "freed", freed) + + remaining := make([]v1alpha1.Reservation, 0, len(allReservations)-1) + for i := range allReservations { + if allReservations[i].Name != target.Name { + remaining = append(remaining, allReservations[i]) + } + } + if postViolations := computeViolations(remaining, hv); len(postViolations) > 0 { + updateMonitor(postViolations) + } else { + monitor.ClearHost(host, az) + } + return true, false, nil +} + +// unplaceReservation clears Spec.TargetHost and Spec.Allocations, and returns the resources +// freed (full Spec.Resources — the slot is fully unplaced). Status cleanup is left to the +// reconcile loop, which will detect the Spec.TargetHost="" / IsReady() mismatch and converge. +func (r *CommitmentReservationController) unplaceReservation( + ctx context.Context, + res *v1alpha1.Reservation, +) (map[hv1.ResourceName]resource.Quantity, error) { + + logger := LoggerFromContext(ctx).WithValues("component", "oversubscription-check", "reservation", res.Name) + + freed := reservations.UnusedReservationCapacity(res, true) + + old := res.DeepCopy() + res.Spec.TargetHost = "" + if res.Spec.CommittedResourceReservation != nil { + res.Spec.CommittedResourceReservation.Allocations = nil + } + logger.Info("unplacing reservation for over-subscription remediation", + "freed", freed, + "previous host", old.Spec.TargetHost, + "flavor", old.Spec.CommittedResourceReservation.ResourceName, + "flavorGroup", old.Spec.CommittedResourceReservation.ResourceGroup, + "hasAllocations", old.Spec.CommittedResourceReservation != nil && len(old.Spec.CommittedResourceReservation.Allocations) > 0) + if err := r.Patch(ctx, res, client.MergeFrom(old)); err != nil { + logger.Error(err, "failed to patch reservation", "reservation", res.Name) + return nil, fmt.Errorf("failed to patch reservation %s: %w", res.Name, err) + } + return freed, nil +} diff --git a/internal/scheduling/reservations/commitments/reservation_controller_monitor.go b/internal/scheduling/reservations/commitments/reservation_controller_monitor.go new file mode 100644 index 000000000..e53104bf7 --- /dev/null +++ b/internal/scheduling/reservations/commitments/reservation_controller_monitor.go @@ -0,0 +1,46 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "github.com/prometheus/client_golang/prometheus" +) + +// ReservationControllerMonitor reports per-host over-subscription violations +// detected by the CommitmentReservationController. +type ReservationControllerMonitor struct { + oversubscribed *prometheus.GaugeVec +} + +func NewReservationControllerMonitor() ReservationControllerMonitor { + return ReservationControllerMonitor{ + oversubscribed: prometheus.NewGaugeVec(prometheus.GaugeOpts{ + Name: "cortex_committed_resource_host_oversubscribed", + Help: "Excess resource units by which a host's reservation blocks + VM allocations exceed its effective capacity. " + + "Non-zero when the host is over-subscribed and unresolvable via unallocated slot eviction. " + + "Transient spikes are expected after live migrations (slot stays on old host until usage reconciler cleans it up).", + }, []string{"host", "az", "resource"}), + } +} + +// SetOversubscribed records the excess amount for a host+resource pair. +// Zero clears the violation. +func (m *ReservationControllerMonitor) SetOversubscribed(host, az, resource string, excessUnits float64) { + m.oversubscribed.WithLabelValues(host, az, resource).Set(excessUnits) +} + +// ClearHost resets all resource gauges for a host that is no longer over-subscribed. +func (m *ReservationControllerMonitor) ClearHost(host, az string) { + m.oversubscribed.DeletePartialMatch(prometheus.Labels{"host": host}) +} + +// Describe implements prometheus.Collector. +func (m *ReservationControllerMonitor) Describe(ch chan<- *prometheus.Desc) { + m.oversubscribed.Describe(ch) +} + +// Collect implements prometheus.Collector. +func (m *ReservationControllerMonitor) Collect(ch chan<- prometheus.Metric) { + m.oversubscribed.Collect(ch) +} diff --git a/internal/scheduling/reservations/commitments/reservation_controller_test.go b/internal/scheduling/reservations/commitments/reservation_controller_test.go index a776ac0cf..df5724363 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller_test.go +++ b/internal/scheduling/reservations/commitments/reservation_controller_test.go @@ -21,6 +21,7 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" schedulerdelegationapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -62,13 +63,15 @@ func TestCommitmentReservationController_Reconcile(t *testing.T) { Name: "test-reservation", }, Spec: v1alpha1.ReservationSpec{ - Type: v1alpha1.ReservationTypeCommittedResource, + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ ProjectID: "test-project", ResourceName: "test-flavor", }, }, Status: v1alpha1.ReservationStatus{ + Host: "host-1", Conditions: []metav1.Condition{ { Type: v1alpha1.ReservationConditionReady, @@ -1216,3 +1219,460 @@ func TestReconcileAllocations_LiveMigration(t *testing.T) { }) } } + +func TestHvCapacityChangePredicate(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + cpu := func(n int64) resource.Quantity { return *resource.NewQuantity(n, resource.DecimalSI) } + + makeHV := func(memGiB, cpuCores int64, instances []hv1.Instance) hv1.Hypervisor { + return hv1.Hypervisor{ + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: gib(memGiB), + hv1.ResourceCPU: cpu(cpuCores), + }, + Instances: instances, + }, + } + } + + tests := []struct { + name string + old hv1.Hypervisor + new hv1.Hypervisor + wantFire bool + }{ + { + name: "instances changed → fires", + old: makeHV(1024, 256, nil), + new: makeHV(1024, 256, []hv1.Instance{{ID: "vm-1"}}), + wantFire: true, + }, + { + name: "effective capacity changed → fires", + old: makeHV(1024, 256, nil), + new: makeHV(2048, 256, nil), + wantFire: true, + }, + { + name: "allocation changed → fires", + old: makeHV(1024, 256, nil), + new: func() hv1.Hypervisor { + h := makeHV(1024, 256, nil) + h.Status.Allocation = map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)} + return h + }(), + wantFire: true, + }, + { + name: "nothing relevant changed → does not fire", + old: makeHV(1024, 256, nil), + new: makeHV(1024, 256, nil), + wantFire: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + oldObj := tt.old + newObj := tt.new + got := hvCapacityChangePredicate.UpdateFunc(event.UpdateEvent{ + ObjectOld: &oldObj, + ObjectNew: &newObj, + }) + if got != tt.wantFire { + t.Errorf("hvCapacityChangePredicate.UpdateFunc = %v, want %v", got, tt.wantFire) + } + }) + } +} + +func TestComputeViolations_NoViolation(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + hv := hv1.Hypervisor{ + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(1024)}, + }, + } + slot := v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + + violations := computeViolations([]v1alpha1.Reservation{slot}, hv) + if len(violations) != 0 { + t.Errorf("expected no violations, got %v", violations) + } +} + +func TestComputeViolations_Oversubscribed(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + hv := hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host-1"}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + }, + } + slot := v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(768)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + + violations := computeViolations([]v1alpha1.Reservation{slot}, hv) + if len(violations) == 0 { + t.Fatal("expected violation for memory") + } + excess := violations[hv1.ResourceMemory] + expected := gib(256) + if excess.Cmp(expected) != 0 { + t.Errorf("expected excess 256 GiB, got %s", excess.String()) + } +} + +func TestCheckHostOversubscription_GracePeriodDefers(t *testing.T) { + scheme := newCRTestScheme(t) + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + hv := hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host-1", Labels: map[string]string{"topology.kubernetes.io/zone": "az1"}}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(1024)}, + }, + } + // 3 x 512 GiB slots on a 1024 GiB host → over-subscribed by 512 GiB + makeSlot := func(name string) v1alpha1.Reservation { + return v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + } + slots := []v1alpha1.Reservation{makeSlot("slot-1"), makeSlot("slot-2"), makeSlot("slot-3")} + + k8sClient := newCRTestClient(scheme, &slots[0], &slots[1], &slots[2]) + monitor := NewReservationControllerMonitor() + controller := &CommitmentReservationController{Client: k8sClient, Monitor: &monitor} + + // First call: no firstSeen yet → grace period starts, no eviction + evicted, resolved, err := controller.checkHostOversubscription(context.Background(), "host-1", + slots, hv, &monitor, 2*time.Minute, time.Time{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if evicted || resolved { + t.Error("expected no eviction and no resolution during initial grace period") + } + + // Second call with elapsed grace period: should evict + firstSeen := time.Now().Add(-3 * time.Minute) + + evicted, _, err = controller.checkHostOversubscription(context.Background(), "host-1", + slots, hv, &monitor, 2*time.Minute, firstSeen) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if !evicted { + t.Error("expected eviction after grace period elapsed") + } + + // Verify the evicted slot has TargetHost cleared + var updated v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: "slot-1"}, &updated); err != nil { + t.Fatalf("failed to get slot: %v", err) + } + if updated.Spec.TargetHost != "" { + t.Errorf("expected TargetHost to be cleared, got %q", updated.Spec.TargetHost) + } +} + +func TestUnplaceReservation_ClearsAllocations(t *testing.T) { + scheme := newCRTestScheme(t) + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + slot := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-1"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-1": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(256)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host-1"}, + }, + }, + } + + k8sClient := newCRTestClient(scheme, slot) + controller := &CommitmentReservationController{Client: k8sClient} + + freed, err := controller.unplaceReservation(context.Background(), slot) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + freedMem := freed[hv1.ResourceMemory] + expected512 := gib(512) + if freedMem.Value() != expected512.Value() { + t.Errorf("expected freed memory = 512 GiB, got %s", freedMem.String()) + } + + var updated v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: "slot-1"}, &updated); err != nil { + t.Fatalf("failed to get updated reservation: %v", err) + } + if updated.Spec.TargetHost != "" { + t.Errorf("expected TargetHost cleared, got %q", updated.Spec.TargetHost) + } + if len(updated.Spec.CommittedResourceReservation.Allocations) != 0 { + t.Errorf("expected Spec.Allocations cleared, got %v", updated.Spec.CommittedResourceReservation.Allocations) + } + // Status cleanup is left to the reconcile loop — not asserted here. +} + +func TestSelectEvictionTarget_PrefersSmallestUnallocated(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + allocated := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-allocated"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-1": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(256)}}, + }, + }, + }, + } + large := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-large"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + } + small := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-small"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(128)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + } + + controller := &CommitmentReservationController{} + violations := map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(128)} + target := controller.selectEvictionTarget(context.Background(), + []v1alpha1.Reservation{allocated, large, small}, violations) + + if target == nil { + t.Fatal("expected a target to be selected") + return + } + if target.Name != "slot-small" { + t.Errorf("expected smallest unallocated slot (slot-small), got %q", target.Name) + } +} + +func TestSelectEvictionTarget_FallsBackToAllocatedWhenNoUnallocated(t *testing.T) { + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + mostly := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-mostly-used"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-1": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(480)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host-1"}, + }, + }, + } + idle := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-idle"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-2": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(10)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-2": "host-1"}, + }, + }, + } + + controller := &CommitmentReservationController{} + violations := map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(128)} + target := controller.selectEvictionTarget(context.Background(), + []v1alpha1.Reservation{mostly, idle}, violations) + + if target == nil { + t.Fatal("expected a target to be selected") + return + } + if target.Name != "slot-idle" { + t.Errorf("expected slot with most unused capacity (slot-idle), got %q", target.Name) + } +} + +func TestRunOversubscriptionCheck_RateLimit(t *testing.T) { + scheme := newCRTestScheme(t) + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + hv := &hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host-1", Labels: map[string]string{"topology.kubernetes.io/zone": "az1"}}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(1024)}, + }, + } + slot := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-1"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + k8sClient := newCRTestClient(scheme, hv, slot) + monitor := NewReservationControllerMonitor() + controller := &CommitmentReservationController{ + Client: k8sClient, + Monitor: &monitor, + Conf: ReservationControllerConfig{ + RequeueIntervalActive: metav1.Duration{Duration: 30 * time.Minute}, + EnableOversubscriptionCheck: true, + }, + } + + // First call: runs the check (no violation, returns 0) + result := controller.runOversubscriptionCheck(context.Background(), "host-1") + if result != 0 { + t.Errorf("expected 0 on first call (no violation), got %v", result) + } + + // Disabled flag: must always return 0 regardless of state. + controller.Conf.EnableOversubscriptionCheck = false + if got := controller.runOversubscriptionCheck(context.Background(), "host-1"); got != 0 { + t.Errorf("expected 0 when check is disabled, got %v", got) + } + controller.Conf.EnableOversubscriptionCheck = true + + // Second call immediately: should be rate-limited, set pending, return remaining interval + result = controller.runOversubscriptionCheck(context.Background(), "host-1") + if result == 0 { + t.Error("expected non-zero requeue when rate-limited") + } + if !controller.oversubscriptionPendingCheck["host-1"] { + t.Error("expected pending flag to be set") + } + + // Third call while already pending: should return 0 (no duplicate requeue) + result = controller.runOversubscriptionCheck(context.Background(), "host-1") + if result != 0 { + t.Errorf("expected 0 when already pending (no duplicate requeue), got %v", result) + } +} + +func TestReconcile_RevokesReadyWhenTargetHostCleared(t *testing.T) { + scheme := newCRTestScheme(t) + + // Slot is Ready (Status.Host set, condition true) but Spec.TargetHost was cleared — + // simulates the oversubscription eviction path. + slot := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-1"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + ResourceName: "test-flavor", + ProjectID: "test-project", + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + Conditions: []metav1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionTrue, + Reason: "ReservationActive", + LastTransitionTime: metav1.Now(), + }, + }, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host-1"}, + }, + }, + } + + k8sClient := newCRTestClient(scheme, slot) + controller := &CommitmentReservationController{ + Client: k8sClient, + Conf: ReservationControllerConfig{RequeueIntervalActive: metav1.Duration{Duration: 30 * time.Minute}}, + } + + _, err := controller.Reconcile(context.Background(), ctrl.Request{NamespacedName: client.ObjectKey{Name: "slot-1"}}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var updated v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: "slot-1"}, &updated); err != nil { + t.Fatalf("failed to get updated reservation: %v", err) + } + if updated.Status.Host != "" { + t.Errorf("expected Status.Host cleared, got %q", updated.Status.Host) + } + if updated.Status.CommittedResourceReservation != nil && len(updated.Status.CommittedResourceReservation.Allocations) != 0 { + t.Errorf("expected Status.Allocations cleared, got %v", updated.Status.CommittedResourceReservation.Allocations) + } + cond := meta.FindStatusCondition(updated.Status.Conditions, v1alpha1.ReservationConditionReady) + if cond == nil || cond.Status != metav1.ConditionFalse { + t.Errorf("expected Ready=False, got %v", cond) + } + if cond != nil && cond.Reason != "PlacementRevoked" { + t.Errorf("expected reason PlacementRevoked, got %q", cond.Reason) + } +} diff --git a/internal/scheduling/reservations/commitments/syncer_monitor_test.go b/internal/scheduling/reservations/commitments/syncer_monitor_test.go index d973a95e9..96d4e173f 100644 --- a/internal/scheduling/reservations/commitments/syncer_monitor_test.go +++ b/internal/scheduling/reservations/commitments/syncer_monitor_test.go @@ -81,6 +81,7 @@ func TestSyncerMonitor_SkipReasonsPreInitialized(t *testing.T) { } if skippedFamily == nil { t.Fatal("cortex_committed_resource_syncer_commitments_skipped_total missing after registration") + return } presentReasons := make(map[string]bool) diff --git a/internal/scheduling/reservations/field_index.go b/internal/scheduling/reservations/field_index.go new file mode 100644 index 000000000..57ef4c222 --- /dev/null +++ b/internal/scheduling/reservations/field_index.go @@ -0,0 +1,56 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package reservations + +import ( + "context" + "errors" + "sync" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/pkg/multicluster" + "sigs.k8s.io/controller-runtime/pkg/client" + logf "sigs.k8s.io/controller-runtime/pkg/log" +) + +// IdxReservationByHost is the field index key for looking up Reservations by host. +// Both Spec.TargetHost and Status.Host are indexed so reservations in transit +// (TargetHost != Status.Host) are found via either field. +// All reservation types are included. +const IdxReservationByHost = "reservations.host" + +var onceIndexReservationByHost sync.Once + +// IndexReservationByHost registers the shared host index on the multicluster client. +// Safe to call multiple times — registration happens only once. +func IndexReservationByHost(ctx context.Context, mcl *multicluster.Client) (err error) { + onceIndexReservationByHost.Do(func() { + log := logf.FromContext(ctx) + err = mcl.IndexField(ctx, + &v1alpha1.Reservation{}, + &v1alpha1.ReservationList{}, + IdxReservationByHost, + func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok { + log.Error(errors.New("unexpected type"), "expected Reservation", "object", obj) + return nil + } + hosts := make(map[string]struct{}) + if res.Spec.TargetHost != "" { + hosts[res.Spec.TargetHost] = struct{}{} + } + if res.Status.Host != "" { + hosts[res.Status.Host] = struct{}{} + } + result := make([]string, 0, len(hosts)) + for h := range hosts { + result = append(result, h) + } + return result + }, + ) + }) + return err +}