From 069aedd75c7958a666ef45abd7f6271f97274bf9 Mon Sep 17 00:00:00 2001 From: Marcel <156897072+mblos@users.noreply.github.com> Date: Fri, 7 Aug 2026 16:31:31 +0200 Subject: [PATCH 1/2] feat: CR controller checks for host overload scenario and resolves via reservation re-placements Signed-off-by: Marcel <156897072+mblos@users.noreply.github.com> --- cmd/manager/main.go | 10 +- .../bundles/cortex-nova/templates/alerts.yaml | 37 +++ helm/bundles/cortex-nova/values.yaml | 5 +- .../reservations/capacity_accounting.go | 39 +++ .../reservations/capacity_accounting_test.go | 184 +++++++++++ .../committed_resource_controller_test.go | 19 ++ .../reservations/commitments/config.go | 3 + .../reservations/commitments/field_index.go | 1 - .../commitments/reservation_controller.go | 296 +++++++++++++++++- .../reservation_controller_monitor.go | 46 +++ .../reservation_controller_test.go | 167 ++++++++++ .../scheduling/reservations/field_index.go | 56 ++++ 12 files changed, 856 insertions(+), 7 deletions(-) create mode 100644 internal/scheduling/reservations/commitments/reservation_controller_monitor.go create mode 100644 internal/scheduling/reservations/field_index.go 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..bdbe5daca 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: | + ( + cortex_kvm_host_capacity_usage{type="utilized"} + + on(compute_host, availability_zone, resource) cortex_kvm_host_capacity_usage{type="reserved"} + + on(compute_host, availability_zone, resource) cortex_kvm_host_capacity_usage{type="failover"} + - on(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 | humanize1024 {{ "}}" }}. + 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..4cb7105c8 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -177,7 +177,7 @@ cortex-scheduling-controllers: "*": "kvm-general-purpose-load-balancing" pipelineDefault: "kvm-general-purpose-load-balancing" # How often to re-verify active Reservation CRDs (healthy state) - requeueIntervalActive: "5m" + requeueIntervalActive: "30m" # Back-off interval when knowledge is unavailable requeueIntervalRetry: "1m" # Back-off interval while a VM allocation is still within allocationGracePeriod @@ -185,6 +185,9 @@ 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: "2m" # 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/reservations/capacity_accounting.go b/internal/scheduling/reservations/capacity_accounting.go index ab305d5d5..fcdffe71c 100644 --- a/internal/scheduling/reservations/capacity_accounting.go +++ b/internal/scheduling/reservations/capacity_accounting.go @@ -75,6 +75,45 @@ 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 != 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/committed_resource_controller_test.go b/internal/scheduling/reservations/commitments/committed_resource_controller_test.go index 293f074a9..28351db04 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() } diff --git a/internal/scheduling/reservations/commitments/config.go b/internal/scheduling/reservations/commitments/config.go index 269ba3e4a..e0645c33e 100644 --- a/internal/scheduling/reservations/commitments/config.go +++ b/internal/scheduling/reservations/commitments/config.go @@ -62,6 +62,9 @@ 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"` // 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/reservation_controller.go b/internal/scheduling/reservations/commitments/reservation_controller.go index f6bf5a385..a4d04911d 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 metric reporting (check still runs, only logging). + 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 @@ -141,6 +155,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 +223,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 } @@ -718,7 +740,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 +833,26 @@ var commitmentReservationPredicate = predicate.Funcs{ }, } +// hvCapacityChangePredicate fires when Status.Instances, Status.Allocation, or +// Status.EffectiveCapacity changes on a Hypervisor. Instances covers VM presence +// (used by allocation verification); Allocation and EffectiveCapacity cover capacity +// accounting (used by the over-subscription check). +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) + }, +} + // 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 +867,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 +893,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 +909,249 @@ 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 { + 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.RequeueIntervalActive.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 + return minCheckInterval - timeSinceLastCheck + time.Second + } else { + // 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 + } + + // 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 { + 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() + return gracePeriod + } + + // Violation detected for the first time — start grace period, requeue after it. + if firstSeen.IsZero() { + r.oversubscriptionFirstSeen[host] = time.Now() + return gracePeriod + } + + // Grace period still running — requeue with remaining time. + if elapsed := time.Since(firstSeen); elapsed < gracePeriod { + return gracePeriod - elapsed + } + + // Grace period elapsed but checkHostOversubscription did not evict (no candidates). + return 0 +} + +// 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"] + + free := reservations.HostFreeCapacity(allReservations, hv) + if free == nil { + return false, false, 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 + } + } + if len(violations) == 0 { + monitor.ClearHost(host, az) + return false, true, nil + } + + for rn, excess := range violations { + monitor.SetOversubscribed(host, az, string(rn), float64(excess.Value())) + } + + if firstSeen.IsZero() { + logger.Info("host over-subscribed, grace period started", "gracePeriod", gracePeriod) + return false, false, nil + } + + elapsed := time.Since(firstSeen) + if elapsed < gracePeriod { + logger.Info("host over-subscribed, waiting grace period", + "elapsed", elapsed.Round(time.Second), + "remaining", (gracePeriod - elapsed).Round(time.Second)) + return false, false, nil + } + + logger.Info("host over-subscribed, evicting one slot", + "violations", func() map[string]string { + m := make(map[string]string, len(violations)) + for rn, q := range violations { + m[string(rn)] = q.String() + } + return m + }()) + + var unallocatedReservations, allocatedReservations []*v1alpha1.Reservation + for i := range allReservations { + res := &allReservations[i] + if res.Spec.Type != v1alpha1.ReservationTypeCommittedResource { + continue + } + if res.Spec.CommittedResourceReservation == nil || + len(res.Spec.CommittedResourceReservation.Allocations) == 0 { + unallocatedReservations = append(unallocatedReservations, res) + } else { + allocatedReservations = append(allocatedReservations, res) + } + } + sort.Slice(unallocatedReservations, func(i, j int) bool { + mi := unallocatedReservations[i].Spec.Resources[hv1.ResourceMemory] + mj := unallocatedReservations[j].Spec.Resources[hv1.ResourceMemory] + return mi.Cmp(mj) < 0 + }) + sort.Slice(allocatedReservations, func(i, j int) bool { + ui := reservations.UnusedReservationCapacity(allocatedReservations[i], false) + uj := reservations.UnusedReservationCapacity(allocatedReservations[j], false) + mi := ui[hv1.ResourceMemory] + mj := uj[hv1.ResourceMemory] + return mi.Cmp(mj) < 0 + }) + memViolation := violations[hv1.ResourceMemory] + for _, res := range allocatedReservations { + unused := reservations.UnusedReservationCapacity(res, false) + unusedMem := unused[hv1.ResourceMemory] + if unusedMem.Cmp(memViolation) >= 0 { + allocatedReservations = []*v1alpha1.Reservation{res} + break + } + } + + candidates := append(unallocatedReservations, allocatedReservations...) + if len(candidates) == 0 { + logger.Error(nil, "host over-subscribed but no CR reservation slots found to evict") + return false, false, nil + } + + target := candidates[0] + freed, err := r.unplaceReservation(ctx, target, host) + if err != nil { + 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", func() map[string]string { + m := make(map[string]string, len(freed)) + for rn, q := range freed { + m[string(rn)] = q.String() + } + return m + }()) + return true, false, nil +} + +// unplaceReservation clears Spec.TargetHost, Spec.Allocations, Status.Host, sets Ready=False, +// and returns the resources freed (full Spec.Resources — the slot is fully unplaced). +func (r *CommitmentReservationController) unplaceReservation( + ctx context.Context, + res *v1alpha1.Reservation, + host string, +) (map[hv1.ResourceName]resource.Quantity, error) { + + freed := reservations.UnusedReservationCapacity(res, true) + + old := res.DeepCopy() + res.Spec.TargetHost = "" + if res.Spec.CommittedResourceReservation != nil { + res.Spec.CommittedResourceReservation.Allocations = nil + } + if err := r.Patch(ctx, res, client.MergeFrom(old)); err != nil { + return nil, fmt.Errorf("failed to patch reservation %s: %w", res.Name, err) + } + if err := r.Get(ctx, client.ObjectKeyFromObject(res), res); err != nil { + return nil, fmt.Errorf("failed to re-fetch reservation %s: %w", res.Name, err) + } + old = res.DeepCopy() + meta.SetStatusCondition(&res.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionFalse, + Reason: "OversubscriptionRemediation", + Message: fmt.Sprintf("evicted from %s due to host over-subscription", host), + }) + res.Status.Host = "" + if res.Status.CommittedResourceReservation != nil { + res.Status.CommittedResourceReservation.Allocations = nil + } + if err := r.Status().Patch(ctx, res, client.MergeFrom(old)); err != nil { + return nil, fmt.Errorf("failed to patch reservation %s status: %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..77e68acb2 --- /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, "az": az}) +} + +// 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..2029c4b43 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" @@ -1216,3 +1217,169 @@ 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 TestCheckHostOversubscription_NoViolation(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, slot) + monitor := NewReservationControllerMonitor() + controller := &CommitmentReservationController{Client: k8sClient, Monitor: &monitor} + + evicted, resolved, err := controller.checkHostOversubscription(context.Background(), "host-1", + []v1alpha1.Reservation{*slot}, hv, &monitor, 2*time.Minute, time.Time{}) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if evicted { + t.Error("expected no eviction when host is not over-subscribed") + } + if !resolved { + t.Error("expected resolved=true when host is not over-subscribed") + } +} + +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) + } +} 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 +} From f5e31e4ec2e218b3af4dc1760ef8731bc8e1b6c6 Mon Sep 17 00:00:00 2001 From: Marcel <156897072+mblos@users.noreply.github.com> Date: Fri, 7 Aug 2026 17:10:20 +0200 Subject: [PATCH 2/2] testing Signed-off-by: Marcel <156897072+mblos@users.noreply.github.com> --- .../commitments/reservation_controller.go | 33 +++- .../reservation_controller_test.go | 187 ++++++++++++++++++ 2 files changed, 213 insertions(+), 7 deletions(-) diff --git a/internal/scheduling/reservations/commitments/reservation_controller.go b/internal/scheduling/reservations/commitments/reservation_controller.go index a4d04911d..bf9861782 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller.go +++ b/internal/scheduling/reservations/commitments/reservation_controller.go @@ -1038,13 +1038,21 @@ func (r *CommitmentReservationController) checkHostOversubscription( } if firstSeen.IsZero() { - logger.Info("host over-subscribed, grace period started", "gracePeriod", gracePeriod) + logger.Info("host over-subscribed, starting grace period", + "gracePeriod", gracePeriod, + "violations", func() map[string]string { + m := make(map[string]string, len(violations)) + for rn, q := range violations { + m[string(rn)] = q.String() + } + return m + }()) return false, false, nil } elapsed := time.Since(firstSeen) if elapsed < gracePeriod { - logger.Info("host over-subscribed, waiting grace period", + logger.V(1).Info("host over-subscribed, grace period in progress", "elapsed", elapsed.Round(time.Second), "remaining", (gracePeriod - elapsed).Round(time.Second)) return false, false, nil @@ -1094,13 +1102,24 @@ func (r *CommitmentReservationController) checkHostOversubscription( } } - candidates := append(unallocatedReservations, allocatedReservations...) - if len(candidates) == 0 { - logger.Error(nil, "host over-subscribed but no CR reservation slots found to evict") + // Pick the eviction target: smallest unallocated first, then smallest allocated. + var target *v1alpha1.Reservation + if len(unallocatedReservations) > 0 { + target = unallocatedReservations[0] + } else if len(allocatedReservations) > 0 { + target = allocatedReservations[0] + } + if target == nil { + logger.Info("host over-subscribed but no evictable CR reservation slots found — manual intervention required", + "violations", func() map[string]string { + m := make(map[string]string, len(violations)) + for rn, q := range violations { + m[string(rn)] = q.String() + } + return m + }()) return false, false, nil } - - target := candidates[0] freed, err := r.unplaceReservation(ctx, target, host) if err != nil { return false, false, err diff --git a/internal/scheduling/reservations/commitments/reservation_controller_test.go b/internal/scheduling/reservations/commitments/reservation_controller_test.go index 2029c4b43..1c19c9caa 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller_test.go +++ b/internal/scheduling/reservations/commitments/reservation_controller_test.go @@ -1383,3 +1383,190 @@ func TestCheckHostOversubscription_GracePeriodDefers(t *testing.T) { 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, "host-1") + 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) + } + 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) + } +} + +func TestCheckHostOversubscription_PrefersUnallocated(t *testing.T) { + scheme := newCRTestScheme(t) + gib := func(n int64) resource.Quantity { return *resource.NewQuantity(n*1024*1024*1024, resource.BinarySI) } + + // Host has 1024 GiB. Two 512 GiB slots + 256 GiB VM allocation = 1280 GiB → over-subscribed by 256 GiB. + 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)}, + Allocation: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(256)}, + }, + } + 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)}}, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + Allocations: map[string]string{"vm-1": "host-1"}, + }, + }, + } + unallocated := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-unallocated"}, + 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"}, + } + // free = 1024 - 256(alloc) - 256(allocated slot remaining) - 512(unallocated) = 0 — exactly at boundary + // Need one more slot to push over. Add a third small unallocated slot. + extra := v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "slot-extra"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, TargetHost: "host-1", + Resources: map[hv1.ResourceName]resource.Quantity{hv1.ResourceMemory: gib(128)}, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{}, + }, + Status: v1alpha1.ReservationStatus{Host: "host-1"}, + } + // free = 1024 - 256(alloc) - 256(allocated remaining) - 512(unallocated) - 128(extra) = -128 GiB + slots := []v1alpha1.Reservation{allocated, unallocated, extra} + + k8sClient := newCRTestClient(scheme, &allocated, &unallocated, &extra) + monitor := NewReservationControllerMonitor() + controller := &CommitmentReservationController{Client: k8sClient, Monitor: &monitor} + + 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.Fatal("expected eviction") + } + // The smallest unallocated slot (extra=128GiB) should be evicted first + var updatedAllocated v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: "slot-allocated"}, &updatedAllocated); err != nil { + t.Fatalf("failed to get slot: %v", err) + } + if updatedAllocated.Spec.TargetHost == "" { + t.Error("allocated slot should not have been evicted") + } + var updatedExtra v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), client.ObjectKey{Name: "slot-extra"}, &updatedExtra); err != nil { + t.Fatalf("failed to get slot: %v", err) + } + if updatedExtra.Spec.TargetHost != "" { + t.Error("smallest unallocated slot should have been evicted") + } +} + +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}}, + } + + // 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) + } + + // 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) + } +}