diff --git a/cmd/manager/main.go b/cmd/manager/main.go index fca0c0550..f69cc9005 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -668,6 +668,20 @@ func main() { os.Exit(1) } } + + if commitmentsConfig.ReservationController.EnableOversubscriptionCheck { + reservationControllerMonitor := commitments.NewReservationControllerMonitor() + metrics.Registry.MustRegister(&reservationControllerMonitor) + + if err := (&commitments.HostOversubscriptionController{ + Client: multiclusterClient, + Conf: commitmentsConfig.ReservationController, + Monitor: &reservationControllerMonitor, + }).SetupWithManager(mgr, multiclusterClient); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "HostOversubscription") + os.Exit(1) + } + } } if slices.Contains(mainConfig.EnabledControllers, "datasource-controllers") { setupLog.Info("enabling controller", "controller", "datasource-controllers") diff --git a/helm/bundles/cortex-nova/templates/alerts.yaml b/helm/bundles/cortex-nova/templates/alerts.yaml index 133d8468a..9e4bfcc01 100644 --- a/helm/bundles/cortex-nova/templates/alerts.yaml +++ b/helm/bundles/cortex-nova/templates/alerts.yaml @@ -749,4 +749,74 @@ 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: 15m + 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 or failover guarantees may not be + satisfied. Common causes: VM migrations with slot not yet reclaimed, outdated/stale reservations, or out of sync issues. + If problem remains, inspect the reservations on this host and check the CR/failover controller logs. + - alert: CortexCommittedResourceHostOversubscribed + expr: max by (host, az, resource) (cortex_committed_resource_host_oversubscribed) > 0 + for: 15m + 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.host {{ "}}" }} CR slots exceed effective capacity for {{ "{{" }} $labels.resource {{ "}}" }}" + description: > + Committed resource reservation slots on host {{ "{{" }} $labels.host {{ "}}" }} exceed + its effective capacity for {{ "{{" }} $labels.resource {{ "}}" }} by {{ "{{" }} $value | humanize {{ "}}" }}. + The oversubscription controller was unable to resolve this by evicting unallocated slots. + Inspect the CR reservations on this host and check the oversubscription controller logs. + - alert: CortexCommittedResourceHostOversubscriptionEvictionsHigh + expr: sum by (az) (increase(cortex_committed_resource_host_oversubscribed_unplaced_reservations_total{service="cortex-nova-metrics"}[1h])) > 10 + 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: "High CR slot eviction rate in AZ {{ "{{" }} $labels.az {{ "}}" }}" + description: > + More than 10 committed resource reservation slots were evicted due to oversubscription + in AZ {{ "{{" }} $labels.az {{ "}}" }} within the last hour ({{ "{{" }} $value | humanize {{ "}}" }} evictions). + This indicates recurring oversubscription events — investigate capacity changes, + concurrent slot creation races, or VM migration patterns in this AZ. + {{- end }} {{- end }} diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 6524c095e..b01cfcaf9 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -185,6 +185,16 @@ 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 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 + enableOversubscriptionCheck: true + # When false, violations are detected and exposed via metrics but no slots are evicted. + enableOversubscriptionUnplaceReservations: 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..6b6c35a42 100644 --- a/internal/scheduling/reservations/capacity_accounting.go +++ b/internal/scheduling/reservations/capacity_accounting.go @@ -25,6 +25,7 @@ import ( // res itself is excluded from step 3 to avoid subtracting its own block from free capacity. // Returns false when the hypervisor has no capacity data. func HostHasCapacityForReservation(allReservations []v1alpha1.Reservation, hv hv1.Hypervisor, res *v1alpha1.Reservation) bool { + // TODO consider refactor with HostFreeCapacity effCap := hv.Status.EffectiveCapacity if effCap == nil { effCap = hv.Status.Capacity @@ -75,6 +76,49 @@ 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 { + // TODO consider refactor with HostHasCapacityForReservation + 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..cb9c7db08 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,9 +138,31 @@ 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() } +func testGiB(n int64) resource.Quantity { + return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) +} + func reconcileReq(name string) ctrl.Request { return ctrl.Request{NamespacedName: types.NamespacedName{Name: name}} } @@ -1195,6 +1218,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..0f926a154 100644 --- a/internal/scheduling/reservations/commitments/config.go +++ b/internal/scheduling/reservations/commitments/config.go @@ -62,6 +62,23 @@ 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 HostOversubscriptionController, which watches + // Hypervisor objects and detects committed-resource oversubscription. + // When false, the controller is not wired up and no oversubscription detection runs. + EnableOversubscriptionCheck bool `json:"enableOversubscriptionCheck,omitempty"` + // EnableOversubscriptionUnplaceReservations controls whether the oversubscription controller + // actively evicts reservation slots when a host is over-subscribed. + // When false, violations are detected and exposed via metrics but no eviction is performed. + // Has no effect when EnableOversubscriptionCheck is false. + EnableOversubscriptionUnplaceReservations bool `json:"enableOversubscriptionUnplaceReservations,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/host_oversubscription_controller.go b/internal/scheduling/reservations/commitments/host_oversubscription_controller.go new file mode 100644 index 000000000..79d08617e --- /dev/null +++ b/internal/scheduling/reservations/commitments/host_oversubscription_controller.go @@ -0,0 +1,400 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "fmt" + "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/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + "github.com/cobaltcore-dev/cortex/pkg/multicluster" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" +) + +// HostOversubscriptionController reconciles Hypervisor objects to detect and remediate +// committed-resource oversubscription. +type HostOversubscriptionController struct { + client.Client + Conf ReservationControllerConfig + Monitor *ReservationControllerMonitor + + // mu protects firstSeen and lastCheck. + // firstSeen tracks when a violation was first detected per host, used to enforce + // the grace period before evicting a reservation. + // lastCheck tracks when a host was last checked, used to rate-limit checks. + // + // NOTE: intentionally kept in memory rather than persisted to an HV annotation. + mu sync.Mutex + firstSeen map[string]time.Time + lastCheck map[string]time.Time +} + +func (r *HostOversubscriptionController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + logger := LoggerFromContext(ctx).WithValues("component", "host-oversubscription", "host", req.Name) + r.mu.Lock() + defer r.mu.Unlock() + host := req.Name + 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 limited, remaining := r.isRateLimited(host, minCheckInterval); limited { + logger.V(1).Info("rate limited", "remaining", remaining) + return ctrl.Result{RequeueAfter: remaining}, nil + } + r.recordCheck(host) + + var hv hv1.Hypervisor + if err := r.Get(ctx, req.NamespacedName, &hv); err != nil { + return ctrl.Result{}, client.IgnoreNotFound(err) + } + + var allReservations v1alpha1.ReservationList + if err := r.List(ctx, &allReservations, client.MatchingFields{reservations.IdxReservationByHost: host}); err != nil { + return ctrl.Result{}, fmt.Errorf("list reservations for host %s: %w", host, err) + } + + requeue, err := r.checkOversubscription(ctx, host, allReservations.Items, hv, gracePeriod) + if err != nil { + return ctrl.Result{}, err + } + if requeue > 0 { + return ctrl.Result{RequeueAfter: requeue}, nil + } + return ctrl.Result{}, nil +} + +// checkOversubscription detects and remediates oversubscription for a single host. +// Returns a requeue duration if the grace period is still running, 0 if resolved or no data. +func (r *HostOversubscriptionController) checkOversubscription( + ctx context.Context, + host string, + allReservations []v1alpha1.Reservation, + hv hv1.Hypervisor, + gracePeriod time.Duration, +) (time.Duration, error) { + + logger := LoggerFromContext(ctx).WithValues("component", "host-oversubscription", "host", host) + az := hv.Labels["topology.kubernetes.io/zone"] + + readyCond := meta.FindStatusCondition(hv.Status.Conditions, hv1.ConditionTypeReady) + if readyCond == nil || readyCond.Status != metav1.ConditionTrue { + logger.V(1).Info("hypervisor not ready, skipping oversubscription check") + return 0, nil + } + + violations := computeViolations(allReservations, hv) + if violations == nil { + logger.V(1).Info("no capacity data for host, skipping oversubscription check") + return 0, nil + } + if len(violations) == 0 { + if !r.firstSeenFor(host).IsZero() { + logger.Info("host oversubscription resolved") + } + r.clearFirstSeen(host) + r.Monitor.ClearHost(host, az) + return 0, nil + } + + r.updateMonitor(host, az, violations) + + firstSeen, isNew := r.getOrSetFirstSeen(host) + if isNew { + logger.Info("oversubscription first detected, grace period starting", + "gracePeriod", gracePeriod, "violations", violations) + } + if elapsed := time.Since(firstSeen); elapsed < gracePeriod { + return gracePeriod - elapsed, nil + } + + logger.Info("grace period elapsed, evicting slot", "violations", violations) + if !r.Conf.EnableOversubscriptionUnplaceReservations { + logger.Info("unplace disabled — violations detected but eviction skipped (enableOversubscriptionUnplaceReservations=false)") + r.resetFirstSeen(host) + return gracePeriod, nil + } + target := selectEvictionTarget(ctx, allReservations, violations) + if target == nil { + logger.Info("no evictable slot found — manual intervention required, will retry after grace period") + r.resetFirstSeen(host) + return gracePeriod, nil + } + + if _, err := unplaceReservation(ctx, r.Client, target); err != nil { + r.updateMonitor(host, az, violations) // keep monitor current on error + return 0, err + } + r.Monitor.IncSlotsUnplaced(az) + + // Recompute violations without the evicted slot to get an accurate post-eviction picture. + 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 { + logger.Info("host still oversubscribed after eviction", "violations", postViolations) + r.updateMonitor(host, az, postViolations) + } else { + logger.Info("host oversubscription resolved after eviction") + r.Monitor.ClearHost(host, az) + } + + r.resetFirstSeen(host) + return gracePeriod, nil +} + +func (r *HostOversubscriptionController) updateMonitor(host, az string, violations map[hv1.ResourceName]resource.Quantity) { + r.Monitor.ClearHost(host, az) + for rn, excess := range violations { + r.Monitor.SetOversubscribed(host, az, string(rn), float64(excess.Value())) + } +} + +func (r *HostOversubscriptionController) isRateLimited(host string, minInterval time.Duration) (bool, time.Duration) { + last := r.lastCheck[host] // safe on nil map — returns zero time + if !last.IsZero() { + if elapsed := time.Since(last); elapsed < minInterval { + return true, minInterval - elapsed + } + } + return false, 0 +} + +func (r *HostOversubscriptionController) recordCheck(host string) { + if r.lastCheck == nil { + r.lastCheck = make(map[string]time.Time) + } + r.lastCheck[host] = time.Now() +} + +func (r *HostOversubscriptionController) firstSeenFor(host string) time.Time { + return r.firstSeen[host] // safe on nil map — returns zero time +} + +// getOrSetFirstSeen returns the first-seen time for the host, setting it to now if not yet recorded. +// The second return value is true when the entry was just created. +func (r *HostOversubscriptionController) getOrSetFirstSeen(host string) (time.Time, bool) { + if t := r.firstSeen[host]; !t.IsZero() { + return t, false + } + r.resetFirstSeen(host) + return r.firstSeen[host], true +} + +func (r *HostOversubscriptionController) clearFirstSeen(host string) { + delete(r.firstSeen, host) // safe on nil map +} + +func (r *HostOversubscriptionController) resetFirstSeen(host string) { + if r.firstSeen == nil { + r.firstSeen = make(map[string]time.Time) + } + r.firstSeen[host] = time.Now() +} + +// SetupWithManager wires the controller into the manager. +// Primary subject: Hypervisor — one reconcile per host, requeue is stable regardless of +// which reservation triggered it. +// Secondary watch: all Reservations mapped to their host via Status.Host — any reservation +// type can affect host capacity (failover slots, VM allocations, etc.), so we watch all. +func (r *HostOversubscriptionController) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client) error { + if err := reservations.IndexReservationByHost(context.Background(), mcl); err != nil { + return fmt.Errorf("failed to set up reservation by host index: %w", err) + } + + enqueueHost := func(res *v1alpha1.Reservation, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + if host := res.Status.Host; host != "" { + q.Add(reconcile.Request{NamespacedName: client.ObjectKey{Name: host}}) + } + } + + reservationToHost := handler.Funcs{ + CreateFunc: func(_ context.Context, e event.CreateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + enqueueHost(e.Object.(*v1alpha1.Reservation), q) + }, + UpdateFunc: func(_ context.Context, e event.UpdateEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + enqueueHost(e.ObjectOld.(*v1alpha1.Reservation), q) // old host — reservation may have moved away + enqueueHost(e.ObjectNew.(*v1alpha1.Reservation), q) + }, + DeleteFunc: func(_ context.Context, e event.DeleteEvent, q workqueue.TypedRateLimitingInterface[reconcile.Request]) { + enqueueHost(e.Object.(*v1alpha1.Reservation), q) + }, + } + + bldr := multicluster.BuildController(mcl, mgr) + var err error + bldr, err = bldr.WatchesMulticluster( + &hv1.Hypervisor{}, + &handler.EnqueueRequestForObject{}, + hvCapacityChangePredicate, + ) + if err != nil { + return err + } + bldr, err = bldr.WatchesMulticluster( + &v1alpha1.Reservation{}, + reservationToHost, + ) + if err != nil { + return err + } + return bldr.Named("host-oversubscription").Complete(r) +} + +// 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 selectEvictionTarget( + ctx context.Context, + allReservations []v1alpha1.Reservation, + violations map[hv1.ResourceName]resource.Quantity, +) *v1alpha1.Reservation { + + logger := LoggerFromContext(ctx).WithValues("component", "host-oversubscription") + var selectedRes *v1alpha1.Reservation + + var unallocated, allocated []*v1alpha1.Reservation + for i := range allReservations { + res := &allReservations[i] + if res.Spec.Type != v1alpha1.ReservationTypeCommittedResource || res.Spec.CommittedResourceReservation == nil { + continue + } + if res.Spec.TargetHost == "" { + continue // already evicted, pending status cleanup + } + 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 { + type allocatedEntry struct { + res *v1alpha1.Reservation + buckets map[hv1.ResourceName]int64 + } + entries := make([]allocatedEntry, 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 + } + } + entries[i] = allocatedEntry{res: res, buckets: buckets} + } + // Sort: 1st by unused mem ratio desc, 2nd by unused CPU ratio desc, 3rd by total mem asc. + sort.SliceStable(entries, func(i, j int) bool { + if d := entries[i].buckets[hv1.ResourceMemory] - entries[j].buckets[hv1.ResourceMemory]; d != 0 { + return d > 0 + } + if d := entries[i].buckets[hv1.ResourceCPU] - entries[j].buckets[hv1.ResourceCPU]; d != 0 { + return d > 0 + } + mi, mj := entries[i].res.Spec.Resources[hv1.ResourceMemory], entries[j].res.Spec.Resources[hv1.ResourceMemory] + return mi.Cmp(mj) < 0 + }) + selectedRes = entries[0].res + } + + 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 +} + +// 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 +func unplaceReservation( + ctx context.Context, + c client.Client, + res *v1alpha1.Reservation, +) (map[hv1.ResourceName]resource.Quantity, error) { + + logger := LoggerFromContext(ctx).WithValues("component", "host-oversubscription", "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", + "freed", freed, + "previous host", old.Spec.TargetHost, + "resources", old.Spec.Resources) + if err := c.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/host_oversubscription_controller_test.go b/internal/scheduling/reservations/commitments/host_oversubscription_controller_test.go new file mode 100644 index 000000000..6cdfa2c03 --- /dev/null +++ b/internal/scheduling/reservations/commitments/host_oversubscription_controller_test.go @@ -0,0 +1,460 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "testing" + "time" + + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" +) + +func hvReadyCondition() metav1.Condition { + return metav1.Condition{ + Type: hv1.ConditionTypeReady, + Status: metav1.ConditionTrue, + Reason: hv1.ConditionReasonReadyReady, + LastTransitionTime: metav1.Now(), + } +} + +func TestComputeViolations(t *testing.T) { + tests := []struct { + name string + hv hv1.Hypervisor + slots []v1alpha1.Reservation + wantNil bool + wantViolation bool + wantExcessGiB int64 + }{ + { + name: "nil capacity returns nil", + hv: hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: "host-1"}}, + wantNil: true, + }, + { + name: "slot within capacity: no violation", + hv: hv1.Hypervisor{ + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: testGiB(1024), + }, + }, + }, + slots: []v1alpha1.Reservation{{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + }}, + }, + { + name: "slot exceeds capacity: violation reported", + hv: hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host-1"}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: testGiB(512), + }, + }, + }, + slots: []v1alpha1.Reservation{{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(768)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + }}, + wantViolation: true, + wantExcessGiB: 256, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + v := computeViolations(tt.slots, tt.hv) + if tt.wantNil { + if v != nil { + t.Errorf("expected nil, got %v", v) + } + return + } + if tt.wantViolation && len(v) == 0 { + t.Fatal("expected violation, got none") + } + if !tt.wantViolation && len(v) != 0 { + t.Errorf("expected no violation, got %v", v) + } + if tt.wantExcessGiB != 0 { + excess := v[hv1.ResourceMemory] + if excess.Cmp(testGiB(tt.wantExcessGiB)) != 0 { + t.Errorf("expected excess %d GiB, got %s", tt.wantExcessGiB, excess.String()) + } + } + }) + } +} + +func TestSelectEvictionTarget(t *testing.T) { + unallocated := func(name string, memGiB int64) 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: testGiB(memGiB)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + } + } + allocated := func(name string, totalGiB, usedGiB int64, vmID 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: testGiB(totalGiB)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + vmID: {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(usedGiB)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{vmID: "host-1"}, + }, + }, + } + } + + violations := map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(128)} + + tests := []struct { + name string + slots []v1alpha1.Reservation + wantName string // empty = expect nil + }{ + { + name: "prefers smallest unallocated over any allocated", + slots: []v1alpha1.Reservation{ + allocated("big-alloc", 512, 256, "vm-1"), + unallocated("large", 512), + unallocated("small", 128), + }, + wantName: "small", + }, + { + name: "3 allocated slots: selects most-idle (validates sort ordering)", + slots: []v1alpha1.Reservation{ + allocated("mostly-used", 512, 480, "vm-1"), // ~6% unused + allocated("half-used", 512, 256, "vm-2"), // ~50% unused + allocated("idle", 512, 10, "vm-3"), // ~98% unused + }, + wantName: "idle", + }, + { + name: "skips non-CR reservation types", + slots: []v1alpha1.Reservation{{ + ObjectMeta: metav1.ObjectMeta{Name: "failover"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeFailover, + TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(512)}, + }, + }}, + wantName: "", + }, + { + name: "skips already-evicted slots (TargetHost cleared)", + slots: []v1alpha1.Reservation{{ + ObjectMeta: metav1.ObjectMeta{Name: "evicted"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + }}, + wantName: "", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + target := selectEvictionTarget(context.Background(), tt.slots, violations) + if tt.wantName == "" { + if target != nil { + t.Errorf("expected nil target, got %q", target.Name) + } + return + } + if target == nil { + t.Fatal("expected a target, got nil") + } + if target.Name != tt.wantName { + t.Errorf("expected %q, got %q", tt.wantName, target.Name) + } + }) + } +} + +func TestCheckOversubscription_StartsGracePeriod(t *testing.T) { + scheme := newCRTestScheme(t) + 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: testGiB(1024)}, + Conditions: []metav1.Condition{hvReadyCondition()}, + }, + } + 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: testGiB(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + } + // 3 × 512 GiB slots on a 1024 GiB host → over-subscribed by 512 GiB + slots := []v1alpha1.Reservation{makeSlot("slot-1"), makeSlot("slot-2"), makeSlot("slot-3")} + k8sClient := newCRTestClient(scheme, &slots[0], &slots[1], &slots[2]) + monitor := NewReservationControllerMonitor() + controller := &HostOversubscriptionController{ + Client: k8sClient, + Monitor: &monitor, + Conf: ReservationControllerConfig{EnableOversubscriptionUnplaceReservations: true}, + } + + requeue, err := controller.checkOversubscription(context.Background(), "host-1", slots, hv, 2*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requeue == 0 { + t.Error("expected non-zero requeue on first violation (grace period)") + } + if controller.firstSeen["host-1"].IsZero() { + t.Error("expected firstSeen recorded after first violation") + } + // No eviction during grace period — all slots must still be placed. + for _, name := range []string{"slot-1", "slot-2", "slot-3"} { + var r v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: name}, &r); err != nil { + t.Fatalf("get %s: %v", name, err) + } + if r.Spec.TargetHost == "" { + t.Errorf("%s: TargetHost cleared during grace period", name) + } + } +} + +func TestCheckOversubscription_EvictsAfterGracePeriod(t *testing.T) { + scheme := newCRTestScheme(t) + 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: testGiB(1024)}, + Conditions: []metav1.Condition{hvReadyCondition()}, + }, + } + 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: testGiB(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 := &HostOversubscriptionController{ + Client: k8sClient, + Monitor: &monitor, + Conf: ReservationControllerConfig{EnableOversubscriptionUnplaceReservations: true}, + firstSeen: map[string]time.Time{"host-1": time.Now().Add(-3 * time.Minute)}, + } + + _, err := controller.checkOversubscription(context.Background(), "host-1", slots, hv, 2*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + var evicted []string + for _, name := range []string{"slot-1", "slot-2", "slot-3"} { + var r v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: name}, &r); err != nil { + t.Fatalf("get %s: %v", name, err) + } + if r.Spec.TargetHost == "" { + evicted = append(evicted, name) + } + } + if len(evicted) != 1 { + t.Errorf("expected exactly 1 slot evicted, got %v", evicted) + } +} + +func TestCheckOversubscription_SkipsWhenHVNotReady(t *testing.T) { + scheme := newCRTestScheme(t) + // HV has capacity data but no Ready condition. + hv := hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{Name: "host-1"}, + Status: hv1.HypervisorStatus{ + EffectiveCapacity: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(512)}, + }, + } + // Slot is oversubscribed relative to HV capacity — should NOT be evicted. + 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: testGiB(768)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + } + k8sClient := newCRTestClient(scheme, &slot) + monitor := NewReservationControllerMonitor() + controller := &HostOversubscriptionController{ + Client: k8sClient, + Monitor: &monitor, + Conf: ReservationControllerConfig{EnableOversubscriptionUnplaceReservations: true}, + } + + requeue, err := controller.checkOversubscription(context.Background(), "host-1", []v1alpha1.Reservation{slot}, hv, 2*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requeue != 0 { + t.Errorf("expected 0 requeue when HV not ready, got %v", requeue) + } + if !controller.firstSeen["host-1"].IsZero() { + t.Error("expected firstSeen not set when HV not ready") + } +} + +func TestCheckOversubscription_NoViolation(t *testing.T) { + scheme := newCRTestScheme(t) + 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: testGiB(1024)}, + Conditions: []metav1.Condition{hvReadyCondition()}, + }, + } + 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: testGiB(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + } + k8sClient := newCRTestClient(scheme, &slot) + monitor := NewReservationControllerMonitor() + controller := &HostOversubscriptionController{ + Client: k8sClient, + Monitor: &monitor, + firstSeen: map[string]time.Time{"host-1": time.Now().Add(-5 * time.Minute)}, + } + + requeue, err := controller.checkOversubscription(context.Background(), "host-1", []v1alpha1.Reservation{slot}, hv, 2*time.Minute) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if requeue != 0 { + t.Errorf("expected 0 requeue when no violation, got %v", requeue) + } + if _, still := controller.firstSeen["host-1"]; still { + t.Error("expected firstSeen cleared after violation resolved") + } +} + +func TestHostOversubscriptionController_RateLimit(t *testing.T) { + controller := &HostOversubscriptionController{} + minInterval := 30 * time.Second + + if limited, _ := controller.isRateLimited("host-1", minInterval); limited { + t.Error("expected not rate-limited before first check") + } + + controller.recordCheck("host-1") + + limited, remaining := controller.isRateLimited("host-1", minInterval) + if !limited { + t.Error("expected rate-limited immediately after recordCheck") + } + if remaining <= 0 || remaining > minInterval { + t.Errorf("remaining out of expected range (0, %v]: %v", minInterval, remaining) + } + + controller.lastCheck["host-1"] = time.Time{} + if limited, _ := controller.isRateLimited("host-1", minInterval); limited { + t.Error("expected not rate-limited after zeroing lastCheck") + } +} + +func TestUnplaceReservation_ClearsAllocations(t *testing.T) { + scheme := newCRTestScheme(t) + 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: testGiB(512)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-1": {Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: testGiB(256)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host-1"}, + }, + }, + } + + k8sClient := newCRTestClient(scheme, slot) + freed, err := unplaceReservation(context.Background(), k8sClient, slot) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + freedMem := freed[hv1.ResourceMemory] + if freedMem.Cmp(testGiB(512)) != 0 { + t.Errorf("expected freed 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("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) + } +} diff --git a/internal/scheduling/reservations/commitments/integration_test.go b/internal/scheduling/reservations/commitments/integration_test.go index ce968b4f4..a0774389d 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" @@ -403,6 +404,7 @@ type intgEnv struct { k8sClient client.Client crController *CommittedResourceController resController *CommitmentReservationController + hvController *HostOversubscriptionController schedulerSrv *httptest.Server } @@ -443,6 +445,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,19 +476,27 @@ 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, + EnableOversubscriptionUnplaceReservations: true, }, } if err := resCtrl.Init(context.Background(), resCtrl.Conf); err != nil { t.Fatalf("resCtrl.Init: %v", err) } - return &intgEnv{k8sClient: k8sClient, crController: crCtrl, resController: resCtrl, schedulerSrv: schedulerSrv} + hvCtrl := &HostOversubscriptionController{ + Client: k8sClient, + Monitor: &monitor, + Conf: resCtrl.Conf, + } + return &intgEnv{k8sClient: k8sClient, crController: crCtrl, resController: resCtrl, hvController: hvCtrl, schedulerSrv: schedulerSrv} } func (e *intgEnv) close() { e.schedulerSrv.Close() } @@ -491,7 +519,18 @@ 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) + } + } +} + +func (e *intgEnv) reconcileHV(t *testing.T, host string) { + t.Helper() + req := ctrl.Request{NamespacedName: types.NamespacedName{Name: host}} + if _, err := e.hvController.Reconcile(context.Background(), req); err != nil { + t.Fatalf("HV reconcile %s: %v", host, err) } } @@ -1010,6 +1049,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 +1224,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 +1272,141 @@ 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), + }, + Conditions: []metav1.Condition{{ + Type: hv1.ConditionTypeReady, + Status: metav1.ConditionTrue, + Reason: hv1.ConditionReasonReadyReady, + LastTransitionTime: metav1.Now(), + }}, + }, + } + + 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 host-1 — triggers oversubscription detection (firstSeen set). + env.reconcileHV(t, "host-1") + if _, detected := env.hvController.firstSeen["host-1"]; !detected { + t.Fatal("expected oversubscription to be detected after first reconcile") + } + + // Drive remediation: fast-forward grace period and reset rate limit each cycle. + resolved := false + for range 25 { + env.hvController.firstSeen["host-1"] = time.Now().Add(-3 * time.Minute) + env.hvController.lastCheck["host-1"] = time.Time{} + env.reconcileHV(t, "host-1") + + if _, still := env.hvController.firstSeen["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..a5fd5cd2d 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller.go +++ b/internal/scheduling/reservations/commitments/reservation_controller.go @@ -6,6 +6,8 @@ package commitments import ( "context" "fmt" + "net/http" + "reflect" "time" "k8s.io/apimachinery/pkg/api/meta" @@ -32,7 +34,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 @@ -114,6 +115,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 @@ -295,7 +318,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 +358,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 +377,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 +748,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 +841,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 +877,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 +903,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 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..d8f4b2b09 --- /dev/null +++ b/internal/scheduling/reservations/commitments/reservation_controller_monitor.go @@ -0,0 +1,58 @@ +// 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 + slotsUnplaced *prometheus.CounterVec +} + +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.", + }, []string{"host", "az", "resource"}), + slotsUnplaced: prometheus.NewCounterVec(prometheus.CounterOpts{ + Name: "cortex_committed_resource_host_oversubscribed_unplaced_reservations_total", + Help: "Total number of reservation slots evicted by the oversubscription controller. " + + "A sustained increase indicates recurring oversubscription events in an AZ.", + }, []string{"az"}), + } +} + +// 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, "az": az}) +} + +// IncSlotsUnplaced increments the counter for slots evicted in the given AZ. +func (m *ReservationControllerMonitor) IncSlotsUnplaced(az string) { + m.slotsUnplaced.WithLabelValues(az).Inc() +} + +// Describe implements prometheus.Collector. +func (m *ReservationControllerMonitor) Describe(ch chan<- *prometheus.Desc) { + m.oversubscribed.Describe(ch) + m.slotsUnplaced.Describe(ch) +} + +// Collect implements prometheus.Collector. +func (m *ReservationControllerMonitor) Collect(ch chan<- prometheus.Metric) { + m.oversubscribed.Collect(ch) + m.slotsUnplaced.Collect(ch) +} diff --git a/internal/scheduling/reservations/commitments/reservation_controller_test.go b/internal/scheduling/reservations/commitments/reservation_controller_test.go index a776ac0cf..98e4cb867 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,131 @@ 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 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 +}