diff --git a/CHANGELOG.md b/CHANGELOG.md index 2d2593024..58c928e83 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,49 @@ # Changelog +## 2026-06-23 — [#966](https://github.com/cobaltcore-dev/cortex/pull/966) + +### cortex v0.1.2 (sha-6daa5050) + +Non-breaking changes: +- Pre-allocate PAYG VMs into CR reservation slots on CR creation/modification ([#951](https://github.com/cobaltcore-dev/cortex/pull/951)) +- Keep failover allocation if VM missing from postgres but present on hypervisor ([#909](https://github.com/cobaltcore-dev/cortex/pull/909)) +- All datasources are synced on restart ([#956](https://github.com/cobaltcore-dev/cortex/pull/956)) +- Honor domain restrictions for CR reservation scheduling ([#955](https://github.com/cobaltcore-dev/cortex/pull/955)) +- Create InFlightReservation as part of Reservation CRD ([#954](https://github.com/cobaltcore-dev/cortex/pull/954)) +- Move committed resource status summary business logic to internal ([#953](https://github.com/cobaltcore-dev/cortex/pull/953)) +- Bypass grace period for confirmed VM departure ([#925](https://github.com/cobaltcore-dev/cortex/pull/925)) +- Track VM placements in reservations and classify no-host-found ([#847](https://github.com/cobaltcore-dev/cortex/pull/847)) +- Refactor reservations: move VMSource to shared package, unify VM data layer ([#930](https://github.com/cobaltcore-dev/cortex/pull/930)) +- Add suffix gX to postgresql ([#939](https://github.com/cobaltcore-dev/cortex/pull/939)) +- Update External dependencies to v1.9.1 ([#952](https://github.com/cobaltcore-dev/cortex/pull/952)), v1.14.46 ([#959](https://github.com/cobaltcore-dev/cortex/pull/959)) +- Update `github.com/sapcc` ([#938](https://github.com/cobaltcore-dev/cortex/pull/938)) + +### cortex-nova v0.0.76 + +Includes updated chart cortex v0.1.2. + +- Add `keystoneSecretRef` and `ssoSecretRef` config keys for domain resolution in committed resource reservation scheduling ([#955](https://github.com/cobaltcore-dev/cortex/pull/955)) + +### cortex-crds v0.0.76 + +Includes updated chart cortex v0.1.2. + +### cortex-cinder v0.0.76 + +Includes updated chart cortex v0.1.2. + +### cortex-pods v0.0.76 + +Includes updated chart cortex v0.1.2. + +### cortex-ironcore v0.0.76 + +Includes updated chart cortex v0.1.2. + +### cortex-manila v0.0.76 + +Includes updated chart cortex v0.1.2. + ## 2026-06-08 — [#919](https://github.com/cobaltcore-dev/cortex/pull/919) ### cortex v0.1.0 (sha-a0373875) diff --git a/api/v1alpha1/reservation_types.go b/api/v1alpha1/reservation_types.go index 4ec3ee9c5..f52797654 100644 --- a/api/v1alpha1/reservation_types.go +++ b/api/v1alpha1/reservation_types.go @@ -21,6 +21,9 @@ const ( ReservationTypeCommittedResource ReservationType = "CommittedResourceReservation" // ReservationTypeFailover is a reservation for failover capacity. ReservationTypeFailover ReservationType = "FailoverReservation" + // ReservationTypeInFlight is a reservation that blocks capacity for virtual + // machines that are currently being scheduled, to avoid double-booking. + ReservationTypeInFlight ReservationType = "InFlightReservation" ) // Label keys for Reservation metadata. @@ -35,6 +38,7 @@ const ( // Reservation type label values ReservationTypeLabelCommittedResource = "committed-resource" ReservationTypeLabelFailover = "failover" + ReservationTypeLabelInFlight = "in-flight" ) // Annotation keys for Reservation metadata. @@ -104,10 +108,30 @@ type FailoverReservationSpec struct { ResourceGroup string `json:"resourceGroup,omitempty"` } +// InFlightReservationSpec defines the nature and shape of the virtual machine +// that is expected to land on the designated reservation slot. This spec +// carries information needed for the scheduler to produce a valid placement +// for new virtual machines for the duration the virtual machine is still +// in buildup. +type InFlightReservationSpec struct { + // VMID is the OpenStack server uuid from Nova assigned to the virtual + // machine expected to land on this reservation slot. + VMID string `json:"vmID,omitempty"` + // UserID is the identifier of the user who owns the virtual machine. + UserID string `json:"userID,omitempty"` + // ProjectID is the identifier of the project/tenant that owns + // the virtual machine. + ProjectID string `json:"projectID,omitempty"` + // Intent defines which kind of virtual machine lifecycle operation + // triggered the placement of this in-flight reservation. + // +kubebuilder:validation:Optional + Intent SchedulingIntent `json:"intent"` +} + // ReservationSpec defines the desired state of Reservation. type ReservationSpec struct { // Type of reservation. - // +kubebuilder:validation:Enum=CommittedResourceReservation;FailoverReservation + // +kubebuilder:validation:Enum=CommittedResourceReservation;FailoverReservation;InFlightReservation // +kubebuilder:validation:Required Type ReservationType `json:"type"` @@ -148,6 +172,10 @@ type ReservationSpec struct { // Only used when Type is FailoverReservation. // +kubebuilder:validation:Optional FailoverReservation *FailoverReservationSpec `json:"failoverReservation,omitempty"` + + // InFlightReservation specifies which kind of virtual machine is expected + // to land on the reserved slot. Set when Type is InFlightReservation. + InFlightReservation *InFlightReservationSpec `json:"inFlightReservation,omitempty"` } const ( @@ -189,6 +217,10 @@ type FailoverReservationStatus struct { AcknowledgedAt *metav1.Time `json:"acknowledgedAt,omitempty"` } +// InFlightReservationStatus defines the status fields specific to +// in-flight reservations. +type InFlightReservationStatus struct{} // No captured state for now. + // ReservationStatus defines the observed state of Reservation. type ReservationStatus struct { // The current status conditions of the reservation. @@ -219,6 +251,11 @@ type ReservationStatus struct { // Only used when Type is FailoverReservation. // +kubebuilder:validation:Optional FailoverReservation *FailoverReservationStatus `json:"failoverReservation,omitempty"` + + // InFlightReservation contains status fields specific to in-flight reservations. + // Only used when Type is InFlightReservation. + // +kubebuilder:validation:Optional + InFlightReservation *InFlightReservationStatus `json:"inFlightReservation,omitempty"` } // +kubebuilder:object:root=true diff --git a/api/v1alpha1/zz_generated.deepcopy.go b/api/v1alpha1/zz_generated.deepcopy.go index a98f0bfff..89fe75d93 100644 --- a/api/v1alpha1/zz_generated.deepcopy.go +++ b/api/v1alpha1/zz_generated.deepcopy.go @@ -1000,6 +1000,36 @@ func (in *IdentityDatasource) DeepCopy() *IdentityDatasource { return out } +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InFlightReservationSpec) DeepCopyInto(out *InFlightReservationSpec) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InFlightReservationSpec. +func (in *InFlightReservationSpec) DeepCopy() *InFlightReservationSpec { + if in == nil { + return nil + } + out := new(InFlightReservationSpec) + in.DeepCopyInto(out) + return out +} + +// DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. +func (in *InFlightReservationStatus) DeepCopyInto(out *InFlightReservationStatus) { + *out = *in +} + +// DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new InFlightReservationStatus. +func (in *InFlightReservationStatus) DeepCopy() *InFlightReservationStatus { + if in == nil { + return nil + } + out := new(InFlightReservationStatus) + in.DeepCopyInto(out) + return out +} + // DeepCopyInto is an autogenerated deepcopy function, copying the receiver, writing into out. in must be non-nil. func (in *KPI) DeepCopyInto(out *KPI) { *out = *in @@ -1778,6 +1808,11 @@ func (in *ReservationSpec) DeepCopyInto(out *ReservationSpec) { *out = new(FailoverReservationSpec) **out = **in } + if in.InFlightReservation != nil { + in, out := &in.InFlightReservation, &out.InFlightReservation + *out = new(InFlightReservationSpec) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReservationSpec. @@ -1810,6 +1845,11 @@ func (in *ReservationStatus) DeepCopyInto(out *ReservationStatus) { *out = new(FailoverReservationStatus) (*in).DeepCopyInto(*out) } + if in.InFlightReservation != nil { + in, out := &in.InFlightReservation, &out.InFlightReservation + *out = new(InFlightReservationStatus) + **out = **in + } } // DeepCopy is an autogenerated deepcopy function, copying the receiver, creating a new ReservationStatus. diff --git a/cmd/manager/main.go b/cmd/manager/main.go index d58be2a29..07f3e9ce2 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -115,6 +115,12 @@ func main() { switch os.Args[1] { case "e2e-nova": novaChecksConfig := conf.GetConfigOrDie[nova.ChecksConfig]() + if len(os.Args) >= 3 { + if err := json.Unmarshal([]byte(os.Args[2]), &novaChecksConfig); err != nil { + slog.Error("invalid json override for e2e-nova", "err", err) + os.Exit(1) + } + } nova.RunChecks(ctx, client, novaChecksConfig) return case "e2e-cinder": @@ -598,10 +604,11 @@ func main() { metrics.Registry.MustRegister(&crControllerMonitor) if err := (&commitments.CommittedResourceController{ - Client: multiclusterClient, - Scheme: mgr.GetScheme(), - Conf: crControllerConf, - Monitor: &crControllerMonitor, + Client: multiclusterClient, + Scheme: mgr.GetScheme(), + Conf: crControllerConf, + Monitor: &crControllerMonitor, + VMSource: commitmentsVMSource, }).SetupWithManager(mgr, multiclusterClient); err != nil { setupLog.Error(err, "unable to create controller", "controller", "CommittedResource") os.Exit(1) diff --git a/go.mod b/go.mod index af17b0ebb..26e32019a 100644 --- a/go.mod +++ b/go.mod @@ -10,7 +10,7 @@ require ( github.com/prometheus/client_golang v1.23.2 github.com/prometheus/client_model v0.6.2 github.com/sapcc/go-bits v0.0.0-20260611141223-328f49772fed - go.xyrillian.de/gg v1.9.0 + go.xyrillian.de/gg v1.10.1 k8s.io/api v0.36.2 k8s.io/apimachinery v0.36.2 k8s.io/client-go v0.36.2 @@ -73,7 +73,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/kylelemons/godebug v1.1.0 // indirect github.com/lib/pq v1.12.3 - github.com/mattn/go-sqlite3 v1.14.45 + github.com/mattn/go-sqlite3 v1.14.47 github.com/moby/sys/user v0.4.0 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect diff --git a/go.sum b/go.sum index df3bf3d36..3ff05f7cf 100644 --- a/go.sum +++ b/go.sum @@ -153,8 +153,8 @@ github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0 github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/lib/pq v1.12.3 h1:tTWxr2YLKwIvK90ZXEw8GP7UFHtcbTtty8zsI+YjrfQ= github.com/lib/pq v1.12.3/go.mod h1:/p+8NSbOcwzAEI7wiMXFlgydTwcgTr3OSKMsD2BitpA= -github.com/mattn/go-sqlite3 v1.14.45 h1:6KA/spDguL3KV8rnybG7ezSaE4SeMR3KC9VbUoAQaIk= -github.com/mattn/go-sqlite3 v1.14.45/go.mod h1:pjEuOr8IwzLJP2MfGeTb0A35jauH+C2kbHKBr7yXKVQ= +github.com/mattn/go-sqlite3 v1.14.47 h1:jOBI62gS7nKeZv+as1oGEy0+1qISgXwH/QBlR6KbfIo= +github.com/mattn/go-sqlite3 v1.14.47/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w= github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0= github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo= github.com/moby/sys/user v0.4.0 h1:jhcMKit7SA80hivmFJcbB1vqmw//wU61Zdui2eQXuMs= @@ -259,8 +259,8 @@ go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= go.uber.org/zap v1.28.0 h1:IZzaP1Fv73/T/pBMLk4VutPl36uNC+OSUh3JLG3FIjo= go.uber.org/zap v1.28.0/go.mod h1:rDLpOi171uODNm/mxFcuYWxDsqWSAVkFdX4XojSKg/Q= -go.xyrillian.de/gg v1.9.0 h1:vszip+UjOBaczo/s9tr6Ij2eo39pxWlVZdbBcLkzXBM= -go.xyrillian.de/gg v1.9.0/go.mod h1:dj+ZhCwC6JKWyFvImhVNXQAErrRcYMUkXu6vwWYNrzQ= +go.xyrillian.de/gg v1.10.1 h1:V6oSU+tl25vaRQaMy6Y3jl/0kNoY/a25x4WIk5zQFAw= +go.xyrillian.de/gg v1.10.1/go.mod h1:DoO4fQSWIrBRlNlCjVyrYM0kAEBt/Jg2GkMH+cGRZ0k= go.yaml.in/yaml/v2 v2.4.4 h1:tuyd0P+2Ont/d6e2rl3be67goVK4R6deVxCUX5vyPaQ= go.yaml.in/yaml/v2 v2.4.4/go.mod h1:gMZqIpDtDqOfM0uNfy0SkpRhvUryYH0Z6wdMYcacYXQ= go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= diff --git a/helm/bundles/cortex-cinder/Chart.yaml b/helm/bundles/cortex-cinder/Chart.yaml index 8b2c53de3..751939f50 100644 --- a/helm/bundles/cortex-cinder/Chart.yaml +++ b/helm/bundles/cortex-cinder/Chart.yaml @@ -5,7 +5,7 @@ apiVersion: v2 name: cortex-cinder description: A Helm chart deploying Cortex for Cinder. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex-postgres @@ -16,12 +16,12 @@ dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-knowledge-controllers # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-scheduling-controllers # Owner info adds a configmap to the kubernetes cluster with information on diff --git a/helm/bundles/cortex-crds/Chart.yaml b/helm/bundles/cortex-crds/Chart.yaml index a27bcd178..8501238a9 100644 --- a/helm/bundles/cortex-crds/Chart.yaml +++ b/helm/bundles/cortex-crds/Chart.yaml @@ -5,13 +5,13 @@ apiVersion: v2 name: cortex-crds description: A Helm chart deploying Cortex CRDs. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 # Owner info adds a configmap to the kubernetes cluster with information on # the service owner. This makes it easier to find out who to contact in case diff --git a/helm/bundles/cortex-ironcore/Chart.yaml b/helm/bundles/cortex-ironcore/Chart.yaml index 0561268b0..7a42ac470 100644 --- a/helm/bundles/cortex-ironcore/Chart.yaml +++ b/helm/bundles/cortex-ironcore/Chart.yaml @@ -5,13 +5,13 @@ apiVersion: v2 name: cortex-ironcore description: A Helm chart deploying Cortex for IronCore. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 # Owner info adds a configmap to the kubernetes cluster with information on # the service owner. This makes it easier to find out who to contact in case diff --git a/helm/bundles/cortex-manila/Chart.yaml b/helm/bundles/cortex-manila/Chart.yaml index 6a54c275b..803146db8 100644 --- a/helm/bundles/cortex-manila/Chart.yaml +++ b/helm/bundles/cortex-manila/Chart.yaml @@ -5,7 +5,7 @@ apiVersion: v2 name: cortex-manila description: A Helm chart deploying Cortex for Manila. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex-postgres @@ -16,12 +16,12 @@ dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-knowledge-controllers # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-scheduling-controllers # Owner info adds a configmap to the kubernetes cluster with information on diff --git a/helm/bundles/cortex-nova/Chart.yaml b/helm/bundles/cortex-nova/Chart.yaml index 5ede9af8d..6dd02a092 100644 --- a/helm/bundles/cortex-nova/Chart.yaml +++ b/helm/bundles/cortex-nova/Chart.yaml @@ -5,7 +5,7 @@ apiVersion: v2 name: cortex-nova description: A Helm chart deploying Cortex for Nova. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex-postgres @@ -16,12 +16,12 @@ dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-knowledge-controllers # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 alias: cortex-scheduling-controllers # Owner info adds a configmap to the kubernetes cluster with information on diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index d7ef28094..a07279ff0 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -179,6 +179,15 @@ cortex-scheduling-controllers: allocationGracePeriod: "15m" # 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 + # domain_name scheduler hint consumed by filter_external_customer. + # Must match the credentials used by the commitments syncer. + keystoneSecretRef: + name: cortex-nova-openstack-keystone + namespace: default + # ssoSecretRef: + # name: cortex-nova-openstack-sso + # namespace: default committedResourceController: # Back-off interval while CommittedResource placement is pending or failed (base for exponential backoff) requeueIntervalRetry: "1m" @@ -188,6 +197,9 @@ cortex-scheduling-controllers: slotCreationDelay: "0ms" # Max Reservation CRDs per CommittedResource on the API path; 0 disables the limit maxSlotsPerCommitment: 0 + # When true, the controller scans the AZ for existing PAYG VMs and pre-allocates them + # into reservation slots before falling back to blind scheduler placement. + enablePaygPreAllocation: false committedResourceAPI: # Timeout for watching CommittedResource CRDs before rolling back watchTimeout: "15s" diff --git a/helm/bundles/cortex-placement-shim/Chart.yaml b/helm/bundles/cortex-placement-shim/Chart.yaml index 096d405af..dc115d3b1 100644 --- a/helm/bundles/cortex-placement-shim/Chart.yaml +++ b/helm/bundles/cortex-placement-shim/Chart.yaml @@ -5,13 +5,13 @@ apiVersion: v2 name: cortex-placement-shim description: A Helm chart deploying the Cortex placement shim. type: application -version: 0.1.1 +version: 0.1.2 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex-shim - name: cortex-shim repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 # Owner info adds a configmap to the kubernetes cluster with information on # the service owner. This makes it easier to find out who to contact in case # of issues. See: https://github.com/sapcc/helm-charts/pkgs/container/helm-charts%2Fowner-info diff --git a/helm/bundles/cortex-pods/Chart.yaml b/helm/bundles/cortex-pods/Chart.yaml index 38ad1d49a..81d5bfce7 100644 --- a/helm/bundles/cortex-pods/Chart.yaml +++ b/helm/bundles/cortex-pods/Chart.yaml @@ -5,13 +5,13 @@ apiVersion: v2 name: cortex-pods description: A Helm chart deploying Cortex for Pods. type: application -version: 0.0.75 +version: 0.0.76 appVersion: 0.1.0 dependencies: # from: file://../../library/cortex - name: cortex repository: oci://ghcr.io/cobaltcore-dev/cortex/charts - version: 0.1.1 + version: 0.1.2 # Owner info adds a configmap to the kubernetes cluster with information on # the service owner. This makes it easier to find out who to contact in case diff --git a/helm/dev/cortex-prometheus-operator/Chart.yaml b/helm/dev/cortex-prometheus-operator/Chart.yaml index 17d417ec2..604fb66ab 100644 --- a/helm/dev/cortex-prometheus-operator/Chart.yaml +++ b/helm/dev/cortex-prometheus-operator/Chart.yaml @@ -10,4 +10,4 @@ dependencies: # CRDs of the prometheus operator, such as PrometheusRule, ServiceMonitor, etc. - name: kube-prometheus-stack repository: oci://ghcr.io/prometheus-community/charts - version: 86.2.3 + version: 86.3.2 diff --git a/helm/library/cortex-shim/Chart.yaml b/helm/library/cortex-shim/Chart.yaml index f93c88648..91c51f728 100644 --- a/helm/library/cortex-shim/Chart.yaml +++ b/helm/library/cortex-shim/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: v2 name: cortex-shim description: A Helm chart to distribute cortex shims. type: application -version: 0.1.1 -appVersion: "sha-a1233a36" +version: 0.1.2 +appVersion: "sha-b7fbbe1a" icon: "https://example.com/icon.png" dependencies: [] diff --git a/helm/library/cortex/Chart.yaml b/helm/library/cortex/Chart.yaml index 162475ed6..826f166ed 100644 --- a/helm/library/cortex/Chart.yaml +++ b/helm/library/cortex/Chart.yaml @@ -2,7 +2,7 @@ apiVersion: v2 name: cortex description: A Helm chart to distribute cortex. type: application -version: 0.1.1 -appVersion: "sha-a1233a36" +version: 0.1.2 +appVersion: "sha-6daa5050" icon: "https://example.com/icon.png" dependencies: [] diff --git a/helm/library/cortex/files/crds/cortex.cloud_reservations.yaml b/helm/library/cortex/files/crds/cortex.cloud_reservations.yaml index b4c5bfa6f..85995dc62 100644 --- a/helm/library/cortex/files/crds/cortex.cloud_reservations.yaml +++ b/helm/library/cortex/files/crds/cortex.cloud_reservations.yaml @@ -173,6 +173,31 @@ spec: (e.g., "hana_medium_v2"). type: string type: object + inFlightReservation: + description: |- + InFlightReservation specifies which kind of virtual machine is expected + to land on the reserved slot. Set when Type is InFlightReservation. + properties: + intent: + description: |- + Intent defines which kind of virtual machine lifecycle operation + triggered the placement of this in-flight reservation. + type: string + projectID: + description: |- + ProjectID is the identifier of the project/tenant that owns + the virtual machine. + type: string + userID: + description: UserID is the identifier of the user who owns the + virtual machine. + type: string + vmID: + description: |- + VMID is the OpenStack server uuid from Nova assigned to the virtual + machine expected to land on this reservation slot. + type: string + type: object resources: additionalProperties: anyOf: @@ -203,6 +228,7 @@ spec: enum: - CommittedResourceReservation - FailoverReservation + - InFlightReservation type: string required: - type @@ -330,6 +356,11 @@ spec: - For Nova: the hypervisor hostname - For Pods: the node name type: string + inFlightReservation: + description: |- + InFlightReservation contains status fields specific to in-flight reservations. + Only used when Type is InFlightReservation. + type: object type: object required: - spec diff --git a/internal/knowledge/datasources/plugins/openstack/controller.go b/internal/knowledge/datasources/plugins/openstack/controller.go index f259dd3fe..96d056a15 100644 --- a/internal/knowledge/datasources/plugins/openstack/controller.go +++ b/internal/knowledge/datasources/plugins/openstack/controller.go @@ -7,6 +7,7 @@ import ( "context" "errors" "net/http" + "sync" "time" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -63,6 +64,10 @@ type OpenStackDatasourceReconciler struct { // Config for the reconciler. conf config + // Tracks datasources that have completed at least one reconcile this process lifetime. + // On first reconcile the timestamp skip is bypassed, so a DB wipe + operator restart + // forces an immediate re-sync of all datasources. + reconciledOnce sync.Map } // Reconcile is part of the main kubernetes reconciliation loop which aims to @@ -91,8 +96,11 @@ func (r *OpenStackDatasourceReconciler) Reconcile(ctx context.Context, req ctrl. return ctrl.Result{}, nil } if datasource.Status.NextSyncTime.After(time.Now()) && datasource.Status.NumberOfObjects != 0 { - log.Info("skipping datasource sync, not yet time", "name", datasource.Name) - return ctrl.Result{RequeueAfter: time.Until(datasource.Status.NextSyncTime.Time)}, nil + if _, seen := r.reconciledOnce.Load(req.NamespacedName); seen { + log.Info("skipping datasource sync, not yet time", "name", datasource.Name) + return ctrl.Result{RequeueAfter: time.Until(datasource.Status.NextSyncTime.Time)}, nil + } + log.Info("first reconcile this process lifetime, forcing sync despite timestamp", "name", datasource.Name) } // Authenticate with the database based on the secret provided in the datasource. @@ -263,6 +271,7 @@ func (r *OpenStackDatasourceReconciler) Reconcile(ctx context.Context, req ctrl. // Calculate the next sync time based on the configured sync interval. log.Info("Finished reconcile", "next", nextTime) + r.reconciledOnce.Store(req.NamespacedName, struct{}{}) return ctrl.Result{RequeueAfter: datasource.Spec.OpenStack.SyncInterval.Duration}, nil } diff --git a/internal/knowledge/datasources/plugins/prometheus/controller.go b/internal/knowledge/datasources/plugins/prometheus/controller.go index 26712b54a..dab9ea3b3 100644 --- a/internal/knowledge/datasources/plugins/prometheus/controller.go +++ b/internal/knowledge/datasources/plugins/prometheus/controller.go @@ -6,6 +6,7 @@ package prometheus import ( "context" "net/http" + "sync" "time" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -50,6 +51,10 @@ type PrometheusDatasourceReconciler struct { conf config // Monitor for tracking the datasource syncs. Monitor datasources.Monitor + // Tracks datasources that have completed at least one reconcile this process lifetime. + // On first reconcile the timestamp skip is bypassed, so a DB wipe + operator restart + // forces an immediate re-sync of all datasources. + reconciledOnce sync.Map } // Reconcile is part of the main kubernetes reconciliation loop which aims to @@ -67,8 +72,11 @@ func (r *PrometheusDatasourceReconciler) Reconcile(ctx context.Context, req ctrl return ctrl.Result{}, nil } if datasource.Status.NextSyncTime.After(time.Now()) && datasource.Status.NumberOfObjects != 0 { - log.Info("skipping datasource sync, not yet time", "name", datasource.Name) - return ctrl.Result{RequeueAfter: time.Until(datasource.Status.NextSyncTime.Time)}, nil + if _, seen := r.reconciledOnce.Load(req.NamespacedName); seen { + log.Info("skipping datasource sync, not yet time", "name", datasource.Name) + return ctrl.Result{RequeueAfter: time.Until(datasource.Status.NextSyncTime.Time)}, nil + } + log.Info("first reconcile this process lifetime, forcing sync despite timestamp", "name", datasource.Name) } newSyncerFunc, ok := supportedMetricSyncers[datasource.Spec.Prometheus.Type] @@ -201,6 +209,7 @@ func (r *PrometheusDatasourceReconciler) Reconcile(ctx context.Context, req ctrl } // Calculate the next sync time based on the configured sync interval. + r.reconciledOnce.Store(req.NamespacedName, struct{}{}) return ctrl.Result{RequeueAfter: time.Until(nextSync)}, nil } diff --git a/internal/scheduling/nova/e2e_checks.go b/internal/scheduling/nova/e2e_checks.go index ad65a57b5..94aaa8642 100644 --- a/internal/scheduling/nova/e2e_checks.go +++ b/internal/scheduling/nova/e2e_checks.go @@ -36,11 +36,36 @@ const ( nRandomRequestsToSend = 50 ) +// ChecksConfig holds configuration for nova e2e checks. type ChecksConfig struct { // Secret ref to keystone credentials stored in a k8s secret. KeystoneSecretRef corev1.SecretReference `json:"keystoneSecretRef"` // Secret ref to SSO credentials stored in a k8s secret, if applicable. SSOSecretRef *corev1.SecretReference `json:"ssoSecretRef"` + // DomainNameHintCheck holds optional configuration for the domain_name hint check. + // When nil the check is skipped. Provide DomainName, EligibleHosts, and FlavorName + // to exercise the filter_external_customer path for CR reservation scheduling. + DomainNameHintCheck *DomainNameHintConfig `json:"domainNameHintCheck,omitempty"` +} + +// DomainNameHintConfig holds parameters for CheckDomainNameHintRouting. +type DomainNameHintConfig struct { + // DomainName is the OpenStack domain name passed as the domain_name scheduler hint. + // Use a real domain name from the target environment so filter_external_customer + // can apply its prefix-matching logic. + DomainName string `json:"domainName"` + // EligibleHosts is the list of compute hostnames offered to the scheduler. + // Include at least one host with CUSTOM_EXTERNAL_CUSTOMER_EXCLUSIVE and one without + // to give the filter something to act on. + EligibleHosts []string `json:"eligibleHosts"` + // FlavorName is the Nova flavor name to use in the request (e.g. "g_k_c1_m2_v2"). + FlavorName string `json:"flavorName"` + // FlavorExtraSpecs are the extra specs for the flavor. Required by most pipeline + // filters to determine hypervisor type, capabilities, and traits. + // Example: {"capabilities:hypervisor_type": "CH", "quota:hw_version": "2101"} + FlavorExtraSpecs map[string]string `json:"flavorExtraSpecs,omitempty"` + // Pipeline is the scheduler pipeline to target. Empty means default. + Pipeline string `json:"pipeline,omitempty"` } // Data necessary to generate a somewhat valid nova scheduler request. @@ -356,6 +381,64 @@ func checkNovaSchedulerReturnsValidHosts( return resp.Hosts } +// CheckDomainNameHintRouting sends a synthetic CR reservation scheduling request +// with _nova_check_type=reserve_for_committed_resource and domain_name=config.DomainName +// to the nova external scheduler and asserts HTTP 200. Confirms the hint flows through +// the pipeline and filter_external_customer evaluates it. Inspect the manager logs to +// see which hosts were kept or dropped. +func CheckDomainNameHintRouting(ctx context.Context, config ChecksConfig) { + cfg := config.DomainNameHintCheck + if cfg == nil { + slog.Info("domain_name hint check skipped: DomainNameHintCheck not configured") + return + } + if cfg.FlavorName == "" || len(cfg.EligibleHosts) == 0 { + slog.Info("domain_name hint check skipped: FlavorName or EligibleHosts not set") + return + } + + hosts := make([]api.ExternalSchedulerHost, len(cfg.EligibleHosts)) + weights := make(map[string]float64, len(cfg.EligibleHosts)) + for i, h := range cfg.EligibleHosts { + hosts[i] = api.ExternalSchedulerHost{ComputeHost: h} + weights[h] = 0.0 + } + + req := api.ExternalSchedulerRequest{ + Pipeline: cfg.Pipeline, + Hosts: hosts, + Weights: weights, + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + InstanceUUID: "e2e-domain-hint-check", + Flavor: api.NovaObject[api.NovaFlavor]{ + Data: api.NovaFlavor{ + Name: cfg.FlavorName, + ExtraSpecs: cfg.FlavorExtraSpecs, + }, + }, + SchedulerHints: map[string]any{ + "_nova_check_type": string(api.ReserveForCommittedResourceIntent), + "domain_name": cfg.DomainName, + }, + }, + }, + } + + slog.Info("domain_name hint check: sending CR reservation scheduling request", + "domainName", cfg.DomainName, + "eligibleHosts", cfg.EligibleHosts, + "flavorName", cfg.FlavorName, + ) + + hosts2 := checkNovaSchedulerReturnsValidHosts(ctx, req) + slog.Info("domain_name hint check passed", + "domainName", cfg.DomainName, + "hostsReturned", len(hosts2), + "hosts", hosts2, + ) +} + // Run all checks. func RunChecks(ctx context.Context, client client.Client, config ChecksConfig) { datacenter := prepare(ctx, client, config) @@ -370,10 +453,10 @@ func RunChecks(ctx context.Context, client client.Client, config ChecksConfig) { requestsWithNoHostsReturned++ } } - // Print a summary. slog.Info( "summary", "requestsWithHostsReturned", requestsWithHostsReturned, "requestsWithNoHostsReturned", requestsWithNoHostsReturned, ) + CheckDomainNameHintRouting(ctx, config) } diff --git a/internal/scheduling/nova/plugins/filters/filter_external_customer_test.go b/internal/scheduling/nova/plugins/filters/filter_external_customer_test.go index 97c9d6925..290709f05 100644 --- a/internal/scheduling/nova/plugins/filters/filter_external_customer_test.go +++ b/internal/scheduling/nova/plugins/filters/filter_external_customer_test.go @@ -63,6 +63,15 @@ func TestFilterExternalCustomerStep_Run(t *testing.T) { Traits: []string{"CUSTOM_EXTERNAL_CUSTOMER_EXCLUSIVE"}, }, }, + &hv1.Hypervisor{ + ObjectMeta: v1.ObjectMeta{ + Name: "host-custom-trait", + }, + Spec: hv1.HypervisorSpec{ + CustomTraits: []string{"CUSTOM_EXTERNAL_CUSTOMER_EXCLUSIVE"}, + }, + // Status.Traits intentionally empty — trait comes solely from Spec.CustomTraits. + }, } tests := []struct { @@ -392,6 +401,50 @@ func TestFilterExternalCustomerStep_Run(t *testing.T) { expectedHosts: []string{"host1", "host3"}, filteredHosts: []string{}, }, + { + name: "ReserveForCommittedResourceIntent with external customer domain - filter applies", + opts: FilterExternalCustomerStepOpts{ + CustomerDomainNamePrefixes: []string{"ext-"}, + }, + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + SchedulerHints: map[string]any{ + "_nova_check_type": string(api.ReserveForCommittedResourceIntent), + "domain_name": "ext-customer1", + }, + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host1"}, + {ComputeHost: "host3"}, + {ComputeHost: "host4"}, + }, + }, + expectedHosts: []string{"host1"}, + filteredHosts: []string{"host3", "host4"}, + }, + { + name: "Trait from Spec.CustomTraits (not Status.Traits) grants host inclusion", + opts: FilterExternalCustomerStepOpts{ + CustomerDomainNamePrefixes: []string{"ext-"}, + }, + request: api.ExternalSchedulerRequest{ + Spec: api.NovaObject[api.NovaSpec]{ + Data: api.NovaSpec{ + SchedulerHints: map[string]any{ + "domain_name": "ext-customer1", + }, + }, + }, + Hosts: []api.ExternalSchedulerHost{ + {ComputeHost: "host-custom-trait"}, + {ComputeHost: "host4"}, + }, + }, + expectedHosts: []string{"host-custom-trait"}, + filteredHosts: []string{"host4"}, + }, } for _, tt := range tests { diff --git a/internal/scheduling/reservations/commitments/committed_resource_controller.go b/internal/scheduling/reservations/commitments/committed_resource_controller.go index a24e0c9e3..cd0bdc88a 100644 --- a/internal/scheduling/reservations/commitments/committed_resource_controller.go +++ b/internal/scheduling/reservations/commitments/committed_resource_controller.go @@ -38,6 +38,9 @@ type CommittedResourceController struct { Scheme *runtime.Scheme Conf CommittedResourceControllerConfig Monitor *CRControllerMonitor + // VMSource enables PAYG pre-allocation when creating reservation slots. When nil the + // PAYG scan is skipped and all slots are created via blind scheduler probes. + VMSource reservations.VMSource } func (r *CommittedResourceController) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { @@ -315,11 +318,16 @@ func (r *CommittedResourceController) applyReservationState(ctx context.Context, state.CreatorRequestID = reservations.GlobalRequestIDFromContext(ctx) state.ParentGeneration = cr.Generation - mgr := NewReservationManager(r.Client) - mgr.SlotCreationDelay = r.Conf.SlotCreationDelay.Duration + maxSlots := 0 if cr.Spec.AllowRejection { - mgr.MaxSlots = r.Conf.MaxSlotsPerCommitment + maxSlots = r.Conf.MaxSlotsPerCommitment } + mgr := NewReservationManager(r.Client, ReservationManagerConfig{ + SlotCreationDelay: r.Conf.SlotCreationDelay.Duration, + MaxSlots: maxSlots, + EnablePaygPreAllocation: r.Conf.EnablePaygPreAllocation, + VMSource: r.VMSource, + }) result, err := mgr.ApplyCommitmentState(ctx, logger, state, flavorGroups, "committed-resource-controller") if err != nil { var limitErr *SlotLimitExceededError @@ -401,7 +409,7 @@ func (r *CommittedResourceController) setAccepted(ctx context.Context, cr *v1alp LastTransitionTime: now, ObservedGeneration: cr.Generation, }) - cr.Status.StatusSummary = v1alpha1.ComputeStatusSummary(cr.Spec, cr.Status, now.Time) + cr.Status.StatusSummary = computeStatusSummary(cr.Spec, cr.Status, now.Time) if err := r.Status().Patch(ctx, cr, client.MergeFrom(old)); err != nil { return client.IgnoreNotFound(err) } @@ -470,7 +478,11 @@ func (r *CommittedResourceController) rollbackToAccepted(ctx context.Context, lo state.NamePrefix = cr.Name + "-" state.CreatorRequestID = reservations.GlobalRequestIDFromContext(ctx) state.ParentGeneration = cr.Generation - if _, err := NewReservationManager(r.Client).ApplyCommitmentState(ctx, logger, state, flavorGroups, "committed-resource-controller-rollback"); err != nil { + rollbackMgr := NewReservationManager(r.Client, ReservationManagerConfig{ + EnablePaygPreAllocation: r.Conf.EnablePaygPreAllocation, + VMSource: r.VMSource, + }) + if _, err := rollbackMgr.ApplyCommitmentState(ctx, logger, state, flavorGroups, "committed-resource-controller-rollback"); err != nil { return fmt.Errorf("rollback apply failed: %w", err) } return nil @@ -529,7 +541,7 @@ func (r *CommittedResourceController) setNotReadyRetry(ctx context.Context, cr * func (r *CommittedResourceController) patchNotReady(ctx context.Context, cr *v1alpha1.CommittedResource, reason, message string, resetTimer bool) error { old := cr.DeepCopy() setReadyConditionFalse(&cr.Status.Conditions, reason, message, cr.Generation, resetTimer) - cr.Status.StatusSummary = v1alpha1.ComputeStatusSummary(cr.Spec, cr.Status, time.Now()) + cr.Status.StatusSummary = computeStatusSummary(cr.Spec, cr.Status, time.Now()) if err := r.Status().Patch(ctx, cr, client.MergeFrom(old)); err != nil { return client.IgnoreNotFound(err) } diff --git a/api/v1alpha1/committed_resource_summary.go b/internal/scheduling/reservations/commitments/committed_resource_summary.go similarity index 81% rename from api/v1alpha1/committed_resource_summary.go rename to internal/scheduling/reservations/commitments/committed_resource_summary.go index 83048f0fa..63fb18443 100644 --- a/api/v1alpha1/committed_resource_summary.go +++ b/internal/scheduling/reservations/commitments/committed_resource_summary.go @@ -1,22 +1,23 @@ // Copyright SAP SE // SPDX-License-Identifier: Apache-2.0 -package v1alpha1 +package commitments import ( "fmt" "strings" "time" + "github.com/cobaltcore-dev/cortex/api/v1alpha1" "k8s.io/apimachinery/pkg/api/meta" ) -// ComputeStatusSummary produces a compact human-readable summary of the committed resource's +// computeStatusSummary produces a compact human-readable summary of the committed resource's // current state for the kubectl wide view. // // Format: {reason}[( diff)] [· {N} VM[s]] [· exp in {duration}|no expiry] -func ComputeStatusSummary(spec CommittedResourceSpec, status CommittedResourceStatus, now time.Time) string { - cond := meta.FindStatusCondition(status.Conditions, CommittedResourceConditionReady) +func computeStatusSummary(spec v1alpha1.CommittedResourceSpec, status v1alpha1.CommittedResourceStatus, now time.Time) string { + cond := meta.FindStatusCondition(status.Conditions, v1alpha1.CommittedResourceConditionReady) if cond == nil { return "" } @@ -37,7 +38,7 @@ func ComputeStatusSummary(spec CommittedResourceSpec, status CommittedResourceSt // generic "waiting for reservation placement" which adds nothing beyond the reason. msg := cond.Message showMsg := msg != "" && msg != "waiting for reservation placement" && - (reason == CommittedResourceReasonRejected || reason == CommittedResourceReasonReserving) + (reason == v1alpha1.CommittedResourceReasonRejected || reason == v1alpha1.CommittedResourceReasonReserving) if showMsg { if len(msg) > 80 { msg = msg[:77] + "..." @@ -46,7 +47,7 @@ func ComputeStatusSummary(spec CommittedResourceSpec, status CommittedResourceSt } // VM count — only meaningful once placement is accepted. - if reason == CommittedResourceReasonAccepted { + if reason == v1alpha1.CommittedResourceReasonAccepted { n := len(status.AssignedInstances) if n == 1 { parts = append(parts, "1 VM") @@ -56,7 +57,7 @@ func ComputeStatusSummary(spec CommittedResourceSpec, status CommittedResourceSt } // Expiry — omit for Rejected (CR is terminal, expiry irrelevant). - if reason != CommittedResourceReasonRejected { + if reason != v1alpha1.CommittedResourceReasonRejected { if spec.EndTime == nil { parts = append(parts, "no expiry") } else if remaining := spec.EndTime.Sub(now); remaining <= 0 { @@ -71,7 +72,7 @@ func ComputeStatusSummary(spec CommittedResourceSpec, status CommittedResourceSt // buildSpecDiff returns a semicolon-separated list of placement-relevant field changes // between spec and the last accepted spec. -func buildSpecDiff(spec CommittedResourceSpec, accepted *CommittedResourceSpec) string { +func buildSpecDiff(spec v1alpha1.CommittedResourceSpec, accepted *v1alpha1.CommittedResourceSpec) string { if accepted == nil { return "" } diff --git a/api/v1alpha1/committed_resource_summary_test.go b/internal/scheduling/reservations/commitments/committed_resource_summary_test.go similarity index 61% rename from api/v1alpha1/committed_resource_summary_test.go rename to internal/scheduling/reservations/commitments/committed_resource_summary_test.go index 13fc13c92..56dec3304 100644 --- a/api/v1alpha1/committed_resource_summary_test.go +++ b/internal/scheduling/reservations/commitments/committed_resource_summary_test.go @@ -1,28 +1,29 @@ // Copyright SAP SE // SPDX-License-Identifier: Apache-2.0 -package v1alpha1 +package commitments import ( "testing" "time" + "github.com/cobaltcore-dev/cortex/api/v1alpha1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) func ptr[T any](v T) *T { return &v } -func makeSpec(amount, fg, az string, endTime *time.Time) CommittedResourceSpec { - s := CommittedResourceSpec{ +func makeSpec(amount, fg, az string, endTime *time.Time) v1alpha1.CommittedResourceSpec { + s := v1alpha1.CommittedResourceSpec{ Amount: resource.MustParse(amount), FlavorGroupName: fg, AvailabilityZone: az, - ResourceType: CommittedResourceTypeMemory, + ResourceType: v1alpha1.CommittedResourceTypeMemory, ProjectID: "proj-1", DomainID: "dom-1", CommitmentUUID: "uuid-1", - State: CommitmentStatusConfirmed, + State: v1alpha1.CommitmentStatusConfirmed, } if endTime != nil { s.EndTime = &metav1.Time{Time: *endTime} @@ -30,7 +31,7 @@ func makeSpec(amount, fg, az string, endTime *time.Time) CommittedResourceSpec { return s } -func makeStatusWithMessage(reason, message string, accepted *CommittedResourceSpec) CommittedResourceStatus { +func makeStatusWithMessage(reason, message string, accepted *v1alpha1.CommittedResourceSpec) v1alpha1.CommittedResourceStatus { status := makeStatus(reason, accepted, nil) if len(status.Conditions) == 0 { panic("makeStatus returned no conditions") @@ -39,19 +40,19 @@ func makeStatusWithMessage(reason, message string, accepted *CommittedResourceSp return status } -func makeStatus(reason string, accepted *CommittedResourceSpec, instances []string) CommittedResourceStatus { - status := CommittedResourceStatus{ +func makeStatus(reason string, accepted *v1alpha1.CommittedResourceSpec, instances []string) v1alpha1.CommittedResourceStatus { + status := v1alpha1.CommittedResourceStatus{ AssignedInstances: instances, } if accepted != nil { status.AcceptedSpec = accepted.DeepCopy() } condStatus := metav1.ConditionTrue - if reason != CommittedResourceReasonAccepted { + if reason != v1alpha1.CommittedResourceReasonAccepted { condStatus = metav1.ConditionFalse } status.Conditions = []metav1.Condition{{ - Type: CommittedResourceConditionReady, + Type: v1alpha1.CommittedResourceConditionReady, Status: condStatus, Reason: reason, }} @@ -64,49 +65,49 @@ func TestComputeStatusSummary(t *testing.T) { tests := []struct { name string - spec CommittedResourceSpec - status CommittedResourceStatus + spec v1alpha1.CommittedResourceSpec + status v1alpha1.CommittedResourceStatus want string }{ { name: "no Ready condition", spec: makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)), - status: CommittedResourceStatus{}, + status: v1alpha1.CommittedResourceStatus{}, want: "", }, { name: "Accepted, 3 VMs, exp in hours", spec: makeSpec("2Gi", "fg1", "az1", in(4*time.Hour+30*time.Minute)), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour+30*time.Minute))), []string{"a", "b", "c"}), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour+30*time.Minute))), []string{"a", "b", "c"}), want: "Accepted · 3 VMs · exp in 4h 30m", }, { name: "Accepted, 1 VM, singular", spec: makeSpec("2Gi", "fg1", "az1", in(25*time.Hour+2*time.Minute)), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(25*time.Hour+2*time.Minute))), []string{"a"}), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(25*time.Hour+2*time.Minute))), []string{"a"}), want: "Accepted · 1 VM · exp in 1d 1h", }, { name: "Accepted, 0 VMs, no expiry", spec: makeSpec("2Gi", "fg1", "az1", nil), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", nil)), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", nil)), nil), want: "Accepted · 0 VMs · no expiry", }, { name: "Accepted with amount diff", spec: makeSpec("5Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour)), - status: makeStatus(CommittedResourceReasonAccepted, + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour))), []string{"a", "b", "c"}), want: "Accepted (amount 2Gi→5Gi) · 3 VMs · exp in 3d 2h", }, { name: "Accepted with az and fg diff", - spec: func() CommittedResourceSpec { + spec: func() v1alpha1.CommittedResourceSpec { s := makeSpec("2Gi", "fg2", "az2", nil) return s }(), - status: makeStatus(CommittedResourceReasonAccepted, + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", nil)), []string{"a"}), want: "Accepted (fg fg1→fg2; az az1→az2) · 1 VM · no expiry", @@ -114,80 +115,80 @@ func TestComputeStatusSummary(t *testing.T) { { name: "Reserving with expiry", spec: makeSpec("2Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour)), - status: makeStatus(CommittedResourceReasonReserving, ptr(makeSpec("2Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour))), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonReserving, ptr(makeSpec("2Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour))), nil), want: "Reserving · exp in 3d 2h", }, { name: "Reserving with amount diff", spec: makeSpec("5Gi", "fg1", "az1", in(time.Hour+3*time.Minute)), - status: makeStatus(CommittedResourceReasonReserving, ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour+3*time.Minute))), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonReserving, ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour+3*time.Minute))), nil), want: "Reserving (amount 2Gi→5Gi) · exp in 1h 3m", }, { name: "Rejected with message", spec: makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)), - status: makeStatusWithMessage(CommittedResourceReasonRejected, "no hosts found for reservation (4/4 slots failed)", ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)))), + status: makeStatusWithMessage(v1alpha1.CommittedResourceReasonRejected, "no hosts found for reservation (4/4 slots failed)", ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)))), want: "Rejected · no hosts found for reservation (4/4 slots failed)", }, { name: "Rejected without message (unchanged)", spec: makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)), - status: makeStatus(CommittedResourceReasonRejected, ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour))), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonRejected, ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour))), nil), want: "Rejected", }, { name: "Reserving with failure message shows it", spec: makeSpec("2Gi", "fg1", "az1", in(time.Hour)), - status: makeStatusWithMessage(CommittedResourceReasonReserving, "no hosts found for reservation (2/4 slots failed)", ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour)))), + status: makeStatusWithMessage(v1alpha1.CommittedResourceReasonReserving, "no hosts found for reservation (2/4 slots failed)", ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour)))), want: "Reserving · no hosts found for reservation (2/4 slots failed) · exp in 1h 0m", }, { name: "Reserving with waiting message skips it", spec: makeSpec("2Gi", "fg1", "az1", in(time.Hour)), - status: makeStatusWithMessage(CommittedResourceReasonReserving, "waiting for reservation placement", ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour)))), + status: makeStatusWithMessage(v1alpha1.CommittedResourceReasonReserving, "waiting for reservation placement", ptr(makeSpec("2Gi", "fg1", "az1", in(time.Hour)))), want: "Reserving · exp in 1h 0m", }, { name: "Rejected with long message truncated", spec: makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)), - status: makeStatusWithMessage(CommittedResourceReasonRejected, "no hosts found for reservation because all hypervisors in this flavor group and availability zone are fully committed and no further capacity exists", ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)))), + status: makeStatusWithMessage(v1alpha1.CommittedResourceReasonRejected, "no hosts found for reservation because all hypervisors in this flavor group and availability zone are fully committed and no further capacity exists", ptr(makeSpec("2Gi", "fg1", "az1", in(4*time.Hour)))), want: "Rejected · no hosts found for reservation because all hypervisors in this flavor group a...", }, { name: "Planned with expiry in days", spec: makeSpec("2Gi", "fg1", "az1", in(3*24*time.Hour+2*time.Hour)), - status: makeStatus(CommittedResourceReasonPlanned, nil, nil), + status: makeStatus(v1alpha1.CommittedResourceReasonPlanned, nil, nil), want: "Planned · exp in 3d 2h", }, { name: "Planned no expiry", spec: makeSpec("2Gi", "fg1", "az1", nil), - status: makeStatus(CommittedResourceReasonPlanned, nil, nil), + status: makeStatus(v1alpha1.CommittedResourceReasonPlanned, nil, nil), want: "Planned · no expiry", }, { name: "expired EndTime shows expired", spec: makeSpec("2Gi", "fg1", "az1", in(-time.Hour)), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(-time.Hour))), []string{"a"}), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(-time.Hour))), []string{"a"}), want: "Accepted · 1 VM · expired", }, { name: "sub-minute expiry shows seconds", spec: makeSpec("2Gi", "fg1", "az1", in(18*time.Second)), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(18*time.Second))), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(18*time.Second))), nil), want: "Accepted · 0 VMs · exp in 18s", }, { name: "sub-hour expiry shows minutes and seconds", spec: makeSpec("2Gi", "fg1", "az1", in(23*time.Minute+5*time.Second)), - status: makeStatus(CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(23*time.Minute+5*time.Second))), nil), + status: makeStatus(v1alpha1.CommittedResourceReasonAccepted, ptr(makeSpec("2Gi", "fg1", "az1", in(23*time.Minute+5*time.Second))), nil), want: "Accepted · 0 VMs · exp in 23m 5s", }, } for _, tc := range tests { t.Run(tc.name, func(t *testing.T) { - got := ComputeStatusSummary(tc.spec, tc.status, now) + got := computeStatusSummary(tc.spec, tc.status, now) if got != tc.want { t.Errorf("got %q, want %q", got, tc.want) } diff --git a/internal/scheduling/reservations/commitments/config.go b/internal/scheduling/reservations/commitments/config.go index 8e18c2f9f..269ba3e4a 100644 --- a/internal/scheduling/reservations/commitments/config.go +++ b/internal/scheduling/reservations/commitments/config.go @@ -8,6 +8,7 @@ import ( "time" "github.com/go-logr/logr" + corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) @@ -67,6 +68,16 @@ type ReservationControllerConfig struct { PipelineDefault string `json:"pipelineDefault"` // FlavorGroupPipelines maps flavor group IDs to pipeline names; "*" acts as catch-all. FlavorGroupPipelines map[string]string `json:"flavorGroupPipelines,omitempty"` + // KeystoneSecretRef references a Kubernetes Secret that holds OpenStack credentials + // used to resolve domain IDs to domain names for the domain_name scheduler hint. + // The secret must contain the same keys as the syncer's keystoneSecretRef. + // When empty, domain name resolution is skipped and filter_external_customer will + // not enforce domain restrictions for CR reservations. + KeystoneSecretRef corev1.SecretReference `json:"keystoneSecretRef,omitempty"` + // SSOSecretRef is an optional reference to a Secret holding SSO credentials. + // Required in environments that use SSO-based Keystone authentication. + // When nil, http.DefaultClient is used, which will fail in SSO-only environments. + SSOSecretRef *corev1.SecretReference `json:"ssoSecretRef,omitempty"` } // CommittedResourceControllerConfig holds tuning knobs for the CommittedResource CRD controller. @@ -91,6 +102,11 @@ type CommittedResourceControllerConfig struct { // Has no effect on the AllowRejection=false (syncer) path. // 0 disables the cap. MaxSlotsPerCommitment int `json:"maxSlotsPerCommitment"` + + // EnablePaygPreAllocation enables scanning the AZ for existing PAYG VMs before creating + // blind reservation slots. When true, the controller absorbs matching PAYG VMs into + // pre-populated slots, consuming CR delta before falling back to the blind scheduler path. + EnablePaygPreAllocation bool `json:"enablePaygPreAllocation,omitempty"` } // ResourceTypeConfig holds per-resource flags for a single resource type within a flavor group. diff --git a/internal/scheduling/reservations/commitments/domain_resolver.go b/internal/scheduling/reservations/commitments/domain_resolver.go new file mode 100644 index 000000000..2d5fc4eab --- /dev/null +++ b/internal/scheduling/reservations/commitments/domain_resolver.go @@ -0,0 +1,65 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "fmt" + "sync" + + "github.com/gophercloud/gophercloud/v2" + "github.com/gophercloud/gophercloud/v2/openstack/identity/v3/domains" +) + +// DomainResolver resolves OpenStack domain IDs to their human-readable names. +// Implementations must be safe for concurrent use. +type DomainResolver interface { + // ResolveDomainName returns the name of the domain with the given ID. + // Returns an error if the domain cannot be found or the lookup fails. + ResolveDomainName(ctx context.Context, domainID string) (string, error) +} + +// keystoneDomainResolver resolves domain IDs via the Keystone identity API. +// Names are cached indefinitely — domain names are immutable for the lifetime +// of an OpenStack deployment, and the controller is a long-lived process. +type keystoneDomainResolver struct { + sc *gophercloud.ServiceClient + mu sync.RWMutex + cache map[string]string // domainID → name +} + +// newKeystoneDomainResolver creates a resolver backed by the given Keystone service client. +// The caller is responsible for authenticating the provider before passing sc here. +func newKeystoneDomainResolver(sc *gophercloud.ServiceClient) *keystoneDomainResolver { + return &keystoneDomainResolver{ + sc: sc, + cache: make(map[string]string), + } +} + +// ResolveDomainName returns the domain name for domainID, fetching it from Keystone on +// first access and serving subsequent calls from the in-process cache. +func (r *keystoneDomainResolver) ResolveDomainName(ctx context.Context, domainID string) (string, error) { + r.mu.RLock() + if name, ok := r.cache[domainID]; ok { + r.mu.RUnlock() + return name, nil + } + r.mu.RUnlock() + + // Upgrade to write-lock. Re-check after acquiring the write-lock to avoid a + // redundant Keystone call when two goroutines race on the same uncached ID. + r.mu.Lock() + defer r.mu.Unlock() + if name, ok := r.cache[domainID]; ok { + return name, nil + } + + domain, err := domains.Get(ctx, r.sc, domainID).Extract() + if err != nil { + return "", fmt.Errorf("keystone: failed to resolve domain %q: %w", domainID, err) + } + r.cache[domainID] = domain.Name + return domain.Name, nil +} diff --git a/internal/scheduling/reservations/commitments/domain_resolver_test.go b/internal/scheduling/reservations/commitments/domain_resolver_test.go new file mode 100644 index 000000000..0e9dd30b9 --- /dev/null +++ b/internal/scheduling/reservations/commitments/domain_resolver_test.go @@ -0,0 +1,190 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + + "github.com/gophercloud/gophercloud/v2" +) + +// newDomainResolverTestServer creates an httptest.Server serving Keystone +// GET /domains/{id} responses. Unknown IDs get 404. +func newDomainResolverTestServer(t *testing.T, domainsByID map[string]string) (*httptest.Server, *atomic.Int32) { + t.Helper() + var callCount atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + const prefix = "/domains/" + if len(r.URL.Path) <= len(prefix) { + http.Error(w, "bad path", http.StatusBadRequest) + return + } + id := r.URL.Path[len(prefix):] + callCount.Add(1) + name, ok := domainsByID[id] + if !ok { + http.Error(w, fmt.Sprintf("domain %q not found", id), http.StatusNotFound) + return + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(map[string]any{"domain": map[string]any{"id": id, "name": name}}); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + })) + return server, &callCount +} + +// newTestServiceClient builds a gophercloud.ServiceClient pointing at serverURL. +func newTestServiceClient(serverURL string) *gophercloud.ServiceClient { + return &gophercloud.ServiceClient{ + ProviderClient: &gophercloud.ProviderClient{}, + Endpoint: serverURL + "/", + Type: "identity", + } +} + +func TestKeystoneDomainResolver(t *testing.T) { + server, callCount := newDomainResolverTestServer(t, map[string]string{ + "domain-a": "alpha", + "domain-b": "beta", + }) + defer server.Close() + resolver := newKeystoneDomainResolver(newTestServiceClient(server.URL)) + + tests := []struct { + name string + domainID string + wantName string + wantErr bool + errContains string + }{ + {name: "resolves domain name", domainID: "domain-a", wantName: "alpha"}, + {name: "resolves second domain independently", domainID: "domain-b", wantName: "beta"}, + {name: "not found returns error", domainID: "nonexistent", wantErr: true}, + {name: "error contains domain ID", domainID: "missing-domain", wantErr: true, errContains: "missing-domain"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + name, err := resolver.ResolveDomainName(context.Background(), tt.domainID) + if tt.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + if tt.errContains != "" && !strings.Contains(err.Error(), tt.errContains) { + t.Errorf("expected error to contain %q, got: %v", tt.errContains, err) + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if name != tt.wantName { + t.Errorf("expected %q, got %q", tt.wantName, name) + } + }) + } + + // Each successful domain ID fetched exactly once; subsequent calls served from cache. + // Error cases (not_found, missing-domain) also hit the server but must not be cached. + // We have 2 success lookups (domain-a, domain-b) + 2 error lookups = 4 total calls. + if n := callCount.Load(); n != 4 { + t.Errorf("expected 4 Keystone calls (2 success + 2 error), got %d", n) + } + + t.Run("cache hit does not re-fetch", func(t *testing.T) { + before := callCount.Load() + if _, err := resolver.ResolveDomainName(context.Background(), "domain-a"); err != nil { + t.Fatalf("unexpected error on cache hit: %v", err) + } + if after := callCount.Load(); after != before { + t.Errorf("expected no additional Keystone calls on cache hit, got %d new call(s)", after-before) + } + }) +} + +func TestKeystoneDomainResolver_ErrorNotCached(t *testing.T) { + // First call fails (5xx); second must retry and succeed — errors must not be cached. + var callCount atomic.Int32 + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if callCount.Add(1) == 1 { + http.Error(w, "internal server error", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(map[string]any{"domain": map[string]any{"id": "domain-flaky", "name": "recovered"}}); err != nil { + http.Error(w, err.Error(), http.StatusInternalServerError) + } + })) + defer server.Close() + resolver := newKeystoneDomainResolver(newTestServiceClient(server.URL)) + + if _, err := resolver.ResolveDomainName(context.Background(), "domain-flaky"); err == nil { + t.Fatal("expected error on first (failing) call, got nil") + } + name, err := resolver.ResolveDomainName(context.Background(), "domain-flaky") + if err != nil { + t.Fatalf("expected success after Keystone recovered, got: %v", err) + } + if name != "recovered" { + t.Errorf("expected %q, got %q", "recovered", name) + } + if callCount.Load() != 2 { + t.Errorf("expected 2 Keystone calls (error not cached), got %d", callCount.Load()) + } +} + +func TestKeystoneDomainResolver_ContextCancelled(t *testing.T) { + blocked := make(chan struct{}) + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-blocked + })) + defer server.Close() + defer close(blocked) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + _, err := newKeystoneDomainResolver(newTestServiceClient(server.URL)).ResolveDomainName(ctx, "domain-x") + if err == nil { + t.Fatal("expected error when context is cancelled, got nil") + } +} + +func TestKeystoneDomainResolver_ConcurrentAccess(t *testing.T) { + server, callCount := newDomainResolverTestServer(t, map[string]string{"domain-c": "shared"}) + defer server.Close() + resolver := newKeystoneDomainResolver(newTestServiceClient(server.URL)) + + const goroutines = 20 + errs := make(chan error, goroutines) + var wg sync.WaitGroup + for range goroutines { + wg.Add(1) + go func() { + defer wg.Done() + name, err := resolver.ResolveDomainName(context.Background(), "domain-c") + if err != nil { + errs <- err + } else if name != "shared" { + errs <- fmt.Errorf("expected %q, got %q", "shared", name) + } + }() + } + wg.Wait() + close(errs) + for err := range errs { + t.Errorf("concurrent call error: %v", err) + } + if n := callCount.Load(); n > int32(goroutines) { + t.Errorf("unexpectedly high Keystone call count: %d", n) + } +} diff --git a/internal/scheduling/reservations/commitments/integration_test.go b/internal/scheduling/reservations/commitments/integration_test.go index 8c51a8162..92fcc6246 100644 --- a/internal/scheduling/reservations/commitments/integration_test.go +++ b/internal/scheduling/reservations/commitments/integration_test.go @@ -29,6 +29,7 @@ import ( schedulerdelegationapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "k8s.io/apimachinery/pkg/api/meta" "k8s.io/apimachinery/pkg/api/resource" @@ -55,6 +56,9 @@ type CRIntegrationTestCase struct { // CRs to create and drive to terminal state. CommittedResources []*v1alpha1.CommittedResource + // When set, the CR controller is given a VMSource for PAYG pre-allocation. + VMSource reservations.VMSource + // When true the mock scheduler returns an empty hosts list (NoHostsFound). SchedulerRejects bool // SchedulerAcceptFirst, when > 0, makes the mock scheduler accept only the first N @@ -63,12 +67,13 @@ type CRIntegrationTestCase struct { SchedulerAcceptFirst int // Expected state after all CRs reach a terminal condition. - ExpectedSlots int // total Reservation CRDs remaining in the store - AcceptedCRs []string // CRs expected Ready=True / Accepted - RejectedCRs []string // CRs expected Ready=False / Rejected - PlannedCRs []string // CRs expected Ready=False / Planned - ExpiredCRs []string // CRs expected Ready=False / Expired - SupersededCRs []string // CRs expected Ready=False / Superseded + ExpectedSlots int // total Reservation CRDs remaining in the store + AcceptedCRs []string // CRs expected Ready=True / Accepted + RejectedCRs []string // CRs expected Ready=False / Rejected + PlannedCRs []string // CRs expected Ready=False / Planned + ExpiredCRs []string // CRs expected Ready=False / Expired + SupersededCRs []string // CRs expected Ready=False / Superseded + ValidateReservations func(t *testing.T, slots []v1alpha1.Reservation) // optional extra assertions } func TestCRIntegration(t *testing.T) { @@ -258,6 +263,70 @@ func TestCRIntegration(t *testing.T) { ExpectedSlots: 0, RejectedCRs: []string{"cr-partial"}, }, + // ------------------------------------------------------------------ + // PAYG pre-allocation + // ------------------------------------------------------------------ + { + // PAYG VM on host-1 matches the CR project + flavor group. + // CR controller pre-allocates the slot; reservation controller marks it + // Ready via the PreAllocated fast-path without calling the scheduler. + Name: "PAYG VM present: slot pre-allocated on HV, no scheduler call needed", + Hypervisors: []*hv1.Hypervisor{ + intgHypervisorWithAZ("host-1", "test-az", "vm-payg-1"), + }, + VMSource: &fakeVMSource{vms: []reservations.VM{{ + UUID: "vm-payg-1", + FlavorName: "test-flavor", + CurrentHypervisor: "host-1", + }}}, + CommittedResources: []*v1alpha1.CommittedResource{ + intgCR("cr-payg", "uuid-intg-payg-1", v1alpha1.CommitmentStatusConfirmed), + }, + SchedulerRejects: true, // scheduler would reject if called — proves it isn't + ExpectedSlots: 1, + AcceptedCRs: []string{"cr-payg"}, + ValidateReservations: func(t *testing.T, slots []v1alpha1.Reservation) { + t.Helper() + if len(slots) != 1 { + t.Fatalf("want 1 slot, got %d", len(slots)) + } + res := slots[0] + if res.Spec.TargetHost != "host-1" { + t.Errorf("TargetHost: want host-1, got %q", res.Spec.TargetHost) + } + if res.Spec.CommittedResourceReservation == nil { + t.Fatal("CommittedResourceReservation is nil") + } + if _, ok := res.Spec.CommittedResourceReservation.Allocations["vm-payg-1"]; !ok { + t.Error("expected vm-payg-1 in Spec.Allocations") + } + }, + }, + { + // No PAYG VMs → falls through to the scheduler. Scheduler accepts → CR accepted. + Name: "no PAYG VMs: falls back to scheduler, CR accepted normally", + Hypervisors: []*hv1.Hypervisor{ + intgHypervisorWithAZ("host-1", "test-az"), + }, + VMSource: &fakeVMSource{vms: nil}, + CommittedResources: []*v1alpha1.CommittedResource{ + intgCR("cr-nopayg", "uuid-intg-payg-2", v1alpha1.CommitmentStatusConfirmed), + }, + ExpectedSlots: 1, + AcceptedCRs: []string{"cr-nopayg"}, + ValidateReservations: func(t *testing.T, slots []v1alpha1.Reservation) { + t.Helper() + if len(slots) != 1 { + t.Fatalf("want 1 slot, got %d", len(slots)) + } + if slots[0].Spec.TargetHost == "" { + t.Error("expected TargetHost set by scheduler (Phase 5 path)") + } + if len(slots[0].Spec.CommittedResourceReservation.Allocations) != 0 { + t.Error("expected no pre-allocations on scheduler-placed slot") + } + }, + }, } for _, tc := range testCases { @@ -290,7 +359,7 @@ func runCRIntegrationTestCase(t *testing.T, tc CRIntegrationTestCase) { objects = append(objects, res) } - env := newIntgEnv(t, objects, schedulerFn) + env := newIntgEnv(t, objects, schedulerFn, tc.VMSource) defer env.close() crNames := make([]string, len(tc.CommittedResources)) @@ -320,6 +389,10 @@ func runCRIntegrationTestCase(t *testing.T, tc CRIntegrationTestCase) { intgAssertCRCondition(t, env.k8sClient, tc.PlannedCRs, metav1.ConditionFalse, v1alpha1.CommittedResourceReasonPlanned) intgAssertCRCondition(t, env.k8sClient, tc.ExpiredCRs, metav1.ConditionFalse, string(v1alpha1.CommitmentStatusExpired)) intgAssertCRCondition(t, env.k8sClient, tc.SupersededCRs, metav1.ConditionFalse, string(v1alpha1.CommitmentStatusSuperseded)) + + if tc.ValidateReservations != nil { + tc.ValidateReservations(t, resList.Items) + } } // ============================================================================ @@ -333,7 +406,7 @@ type intgEnv struct { schedulerSrv *httptest.Server } -func newIntgEnv(t *testing.T, initialObjects []client.Object, schedulerFn http.HandlerFunc) *intgEnv { +func newIntgEnv(t *testing.T, initialObjects []client.Object, schedulerFn http.HandlerFunc, vmSource reservations.VMSource) *intgEnv { t.Helper() scheme := newCRTestScheme(t) @@ -377,7 +450,11 @@ func newIntgEnv(t *testing.T, initialObjects []client.Object, schedulerFn http.H crCtrl := &CommittedResourceController{ Client: k8sClient, Scheme: scheme, - Conf: CommittedResourceControllerConfig{RequeueIntervalRetry: metav1.Duration{Duration: 5 * time.Minute}}, + Conf: CommittedResourceControllerConfig{ + RequeueIntervalRetry: metav1.Duration{Duration: 5 * time.Minute}, + EnablePaygPreAllocation: vmSource != nil, + }, + VMSource: vmSource, } resCtrl := &CommitmentReservationController{ Client: k8sClient, @@ -399,7 +476,7 @@ func (e *intgEnv) close() { e.schedulerSrv.Close() } func newDefaultIntgEnv(t *testing.T) *intgEnv { t.Helper() objects := []client.Object{newTestFlavorKnowledge(), intgHypervisor("host-1")} - return newIntgEnv(t, objects, intgAcceptScheduler) + return newIntgEnv(t, objects, intgAcceptScheduler, nil) } func (e *intgEnv) reconcileCR(t *testing.T, crName string) { @@ -630,6 +707,21 @@ func intgHypervisor(name string) *hv1.Hypervisor { return &hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: name}} } +// intgHypervisorWithAZ returns a Hypervisor in the given AZ with optional active instances. +func intgHypervisorWithAZ(name, az string, instanceIDs ...string) *hv1.Hypervisor { + instances := make([]hv1.Instance, len(instanceIDs)) + for i, id := range instanceIDs { + instances[i] = hv1.Instance{ID: id, Name: id, Active: true} + } + return &hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{"topology.kubernetes.io/zone": az}, + }, + Status: hv1.HypervisorStatus{Instances: instances}, + } +} + // intgCR returns a CommittedResource with the default 4 GiB amount. // commitmentUUID must be unique per test case to avoid field-index collisions. func intgCR(name, commitmentUUID string, state v1alpha1.CommitmentStatus) *v1alpha1.CommittedResource { @@ -886,7 +978,7 @@ func TestCRLifecycle(t *testing.T) { }) t.Run("AllowRejection=false: stays Reserving when scheduler rejects", func(t *testing.T) { - env := newIntgEnv(t, []client.Object{newTestFlavorKnowledge(), intgHypervisor("host-1")}, intgRejectScheduler) + env := newIntgEnv(t, []client.Object{newTestFlavorKnowledge(), intgHypervisor("host-1")}, intgRejectScheduler, nil) defer env.close() cr := newTestCommittedResource("my-cr", v1alpha1.CommitmentStatusConfirmed) @@ -995,7 +1087,7 @@ func TestCRLifecycle(t *testing.T) { t.Run("resize failure: rolls back to AcceptedSpec, prior slot preserved", func(t *testing.T) { // Scheduler: accepts the first placement call (initial 4 GiB slot), rejects all subsequent. objects := []client.Object{newTestFlavorKnowledge(), intgHypervisor("host-1")} - env := newIntgEnv(t, objects, intgAcceptFirstScheduler(1)) + env := newIntgEnv(t, objects, intgAcceptFirstScheduler(1), nil) defer env.close() cr := intgCRAllowRejection("my-cr", "uuid-resize-0001", v1alpha1.CommitmentStatusConfirmed) @@ -1060,7 +1152,7 @@ func TestCRLifecycle(t *testing.T) { // then accepts all subsequent. AllowRejection=false means the CR controller retries rather // than rejecting, so the CR must eventually reach Accepted once the scheduler cooperates. objects := []client.Object{newTestFlavorKnowledge(), intgHypervisor("host-1")} - env := newIntgEnv(t, objects, intgRejectFirstScheduler(2)) + env := newIntgEnv(t, objects, intgRejectFirstScheduler(2), nil) defer env.close() cr := newTestCommittedResource("my-cr", v1alpha1.CommitmentStatusConfirmed) diff --git a/internal/scheduling/reservations/commitments/payg_candidates.go b/internal/scheduling/reservations/commitments/payg_candidates.go new file mode 100644 index 000000000..7894852b5 --- /dev/null +++ b/internal/scheduling/reservations/commitments/payg_candidates.go @@ -0,0 +1,171 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "sort" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins/compute" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// PAYGCandidate is an unallocated PAYG VM suitable for pre-allocation into a reservation slot. +type PAYGCandidate struct { + VMID string + HVName string + MemoryMB uint64 + FlavorName string // exact flavor of the VM — used directly as the slot ResourceName +} + +// ScanAZForPaygCandidates returns unallocated PAYG VMs across all HVs in az, grouped by HV name. +// Makes exactly two calls regardless of HV or VM count: +// 1. List all CR Reservations → build allocated VM UUID set from Spec.Allocations +// 2. VMSource.ListVMsByProject → enriched VM data for the project +// +// HVs are identified by the label "topology.kubernetes.io/zone". +func ScanAZForPaygCandidates( + ctx context.Context, + k8sClient client.Client, + vmSource reservations.VMSource, + az string, + projectID string, + flavorGroup compute.FlavorGroupFeature, +) (map[string][]PAYGCandidate, error) { + + var hvList hv1.HypervisorList + if err := k8sClient.List(ctx, &hvList); err != nil { + return nil, err + } + azHVs := make(map[string]*hv1.Hypervisor) + for i := range hvList.Items { + hv := &hvList.Items[i] + if hv.Labels["topology.kubernetes.io/zone"] == az { + azHVs[hv.Name] = hv + } + } + if len(azHVs) == 0 { + return nil, nil + } + + // One cache scan: build the set of VM UUIDs already claimed by any CR Reservation. + // Spec.Allocations is the scheduling perspective — exactly what we need here. + allocatedVMIDs, err := buildAllocatedVMSet(ctx, k8sClient, az) + if err != nil { + return nil, err + } + + projectVMs, err := vmSource.ListVMsByProject(ctx, projectID) + if err != nil { + return nil, err + } + + vmsByHV := make(map[string][]reservations.VM, len(azHVs)) + for _, vm := range projectVMs { + if _, inAZ := azHVs[vm.CurrentHypervisor]; inAZ { + vmsByHV[vm.CurrentHypervisor] = append(vmsByHV[vm.CurrentHypervisor], vm) + } + } + + result := make(map[string][]PAYGCandidate) + for hvName, hv := range azHVs { + candidates := filterPaygCandidates(hvName, hv, vmsByHV[hvName], flavorGroup, allocatedVMIDs) + if len(candidates) > 0 { + result[hvName] = candidates + } + } + return result, nil +} + +// buildAllocatedVMSet lists CR Reservations in az and returns the set of VM UUIDs present +// in any Spec.Allocations. Filtering by AZ keeps the set small — only slots relevant +// to the current scan are included. One cache scan shared across all HV filters. +func buildAllocatedVMSet(ctx context.Context, k8sClient client.Client, az string) (map[string]struct{}, error) { + var resList v1alpha1.ReservationList + if err := k8sClient.List(ctx, &resList, + client.MatchingLabels{v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource}, + ); err != nil { + return nil, err + } + allocated := make(map[string]struct{}) + for _, res := range resList.Items { + if res.Spec.AvailabilityZone != az { + continue + } + if res.Spec.CommittedResourceReservation == nil { + continue + } + for vmUUID := range res.Spec.CommittedResourceReservation.Allocations { + allocated[vmUUID] = struct{}{} + } + } + return allocated, nil +} + +// filterPaygCandidates returns unallocated PAYG VMs from a pre-fetched list for one HV. +// vms must already be filtered to hvName by the caller. +// allocatedVMIDs is a pre-built set of UUIDs already claimed by any CR Reservation.Spec.Allocations. +// Used by reuse sites (ticket #410, #372) where the caller already holds VM data and the allocated set. +func filterPaygCandidates( + hvName string, + hv *hv1.Hypervisor, + vms []reservations.VM, + flavorGroup compute.FlavorGroupFeature, + allocatedVMIDs map[string]struct{}, +) []PAYGCandidate { + + if len(vms) == 0 { + return nil + } + + // Build set of active instance UUIDs from the HV CRD for physical-presence check. + activeOnHV := make(map[string]bool, len(hv.Status.Instances)) + for _, inst := range hv.Status.Instances { + if inst.Active { + activeOnHV[inst.ID] = true + } + } + + // Build set of flavor names in the group for O(1) membership check. + flavorNames := make(map[string]uint64, len(flavorGroup.Flavors)) + for _, f := range flavorGroup.Flavors { + flavorNames[f.Name] = f.MemoryMB + } + + var candidates []PAYGCandidate + for _, vm := range vms { + memMB, inGroup := flavorNames[vm.FlavorName] + if !inGroup { + continue + } + if !activeOnHV[vm.UUID] { + continue + } + if _, isAllocated := allocatedVMIDs[vm.UUID]; isAllocated { + continue + } + candidates = append(candidates, PAYGCandidate{ + VMID: vm.UUID, + HVName: hvName, + MemoryMB: memMB, + FlavorName: vm.FlavorName, + }) + } + + sortCandidatesDesc(candidates) + return candidates +} + +// sortCandidatesDesc sorts candidates descending by memory, with UUID as a stable tie-break. +func sortCandidatesDesc(candidates []PAYGCandidate) { + sort.Slice(candidates, func(i, j int) bool { + if candidates[i].MemoryMB != candidates[j].MemoryMB { + return candidates[i].MemoryMB > candidates[j].MemoryMB + } + return candidates[i].VMID < candidates[j].VMID + }) +} diff --git a/internal/scheduling/reservations/commitments/payg_candidates_test.go b/internal/scheduling/reservations/commitments/payg_candidates_test.go new file mode 100644 index 000000000..a607d51cf --- /dev/null +++ b/internal/scheduling/reservations/commitments/payg_candidates_test.go @@ -0,0 +1,324 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "errors" + "testing" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +// ============================================================================ +// Fake VMSource +// ============================================================================ + +type fakeVMSource struct { + vms []reservations.VM + err error +} + +func (f *fakeVMSource) ListVMs(_ context.Context) ([]reservations.VM, error) { + return f.vms, f.err +} +func (f *fakeVMSource) ListVMsByProject(_ context.Context, _ string) ([]reservations.VM, error) { + return f.vms, f.err +} +func (f *fakeVMSource) ListVMsOnHypervisors(_ context.Context, _ *hv1.HypervisorList, _ bool) ([]reservations.VM, error) { + return f.vms, f.err +} +func (f *fakeVMSource) GetVM(_ context.Context, _ string) (*reservations.VM, error) { + return nil, nil +} +func (f *fakeVMSource) IsServerActive(_ context.Context, _ string) (bool, error) { + return false, nil +} +func (f *fakeVMSource) GetDeletedVMInfo(_ context.Context, _ string) (*reservations.DeletedVMInfo, error) { + return nil, nil +} + +// ============================================================================ +// Helpers +// ============================================================================ + +func paygScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := v1alpha1.AddToScheme(s); err != nil { + t.Fatalf("add v1alpha1 scheme: %v", err) + } + if err := hv1.AddToScheme(s); err != nil { + t.Fatalf("add hv1 scheme: %v", err) + } + return s +} + +func hvWithInstances(name, az string, instances ...hv1.Instance) *hv1.Hypervisor { //nolint:unparam + return &hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{"topology.kubernetes.io/zone": az}, + }, + Status: hv1.HypervisorStatus{Instances: instances}, + } +} + +func activeInstance(id string) hv1.Instance { + return hv1.Instance{ID: id, Name: id, Active: true} +} + +func inactiveInstance(id string) hv1.Instance { + return hv1.Instance{ID: id, Name: id, Active: false} +} + +func vmOnHV(uuid, hvName, flavorName string, memMB uint64) reservations.VM { //nolint:unparam + return reservations.VM{ + UUID: uuid, + FlavorName: flavorName, + CurrentHypervisor: hvName, + Resources: map[string]resource.Quantity{ + "memory": *resource.NewQuantity(int64(memMB)*1024*1024, resource.BinarySI), //nolint:gosec + }, + } +} + +func reservationWithAlloc(name, vmUUID string) *v1alpha1.Reservation { + return &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + CommitmentUUID: "other-cr", + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + vmUUID: {CreationTimestamp: metav1.Now()}, + }, + }, + }, + } +} + +func buildPaygClient(t *testing.T, objs ...client.Object) client.Client { + t.Helper() + return fake.NewClientBuilder(). + WithScheme(paygScheme(t)). + WithObjects(objs...). + WithIndex(&v1alpha1.Reservation{}, idxReservationByAllocationVMUUID, func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok || res.Spec.CommittedResourceReservation == nil { + return nil + } + var uuids []string + for vmUUID := range res.Spec.CommittedResourceReservation.Allocations { + uuids = append(uuids, vmUUID) + } + return uuids + }). + Build() +} + +// ============================================================================ +// Tests: filterPaygCandidates +// ============================================================================ + +func TestFilterPaygCandidates(t *testing.T) { + fg := testFlavorGroup() // flavors: large=32GiB, medium=16GiB, small=8GiB + + tests := []struct { + name string + hv *hv1.Hypervisor + vms []reservations.VM + extraObjs []*v1alpha1.Reservation // reservations with pre-existing allocations + wantVMIDs []string // expected candidate UUIDs in order (largest-first) + wantCount int + }{ + { + name: "VM matches project and flavor group, not allocated — included", + hv: hvWithInstances("host-1", "az-1", activeInstance("vm-a")), + vms: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + wantVMIDs: []string{"vm-a"}, + }, + { + name: "VM already in Reservation.Spec.Allocations — excluded", + hv: hvWithInstances("host-1", "az-1", activeInstance("vm-a")), + vms: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + extraObjs: []*v1alpha1.Reservation{reservationWithAlloc("res-1", "vm-a")}, + wantCount: 0, + }, + { + name: "VM flavor not in flavor group — excluded", + hv: hvWithInstances("host-1", "az-1", activeInstance("vm-a")), + vms: []reservations.VM{vmOnHV("vm-a", "host-1", "unknown-flavor", 8192)}, + wantCount: 0, + }, + { + name: "VM not active in HV CRD Status.Instances — excluded", + hv: hvWithInstances("host-1", "az-1", inactiveInstance("vm-a")), + vms: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + wantCount: 0, + }, + { + name: "VM not present in HV CRD instances at all — excluded", + hv: hvWithInstances("host-1", "az-1"), // no instances + vms: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + wantCount: 0, + }, + { + name: "empty VM list — returns empty, no error", + hv: hvWithInstances("host-1", "az-1", activeInstance("vm-a")), + vms: nil, + wantCount: 0, + }, + { + name: "multiple VMs — unallocated included, allocated excluded, sorted descending by memory", + hv: hvWithInstances("host-1", "az-1", + activeInstance("vm-small"), + activeInstance("vm-medium"), + activeInstance("vm-large"), + activeInstance("vm-allocated"), + ), + vms: []reservations.VM{ + vmOnHV("vm-small", "host-1", "small", 8192), + vmOnHV("vm-medium", "host-1", "medium", 16384), + vmOnHV("vm-large", "host-1", "large", 32768), + vmOnHV("vm-allocated", "host-1", "small", 8192), + }, + extraObjs: []*v1alpha1.Reservation{reservationWithAlloc("res-1", "vm-allocated")}, + wantVMIDs: []string{"vm-large", "vm-medium", "vm-small"}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + allocatedVMIDs := make(map[string]struct{}) + for _, res := range tc.extraObjs { + for vmUUID := range res.Spec.CommittedResourceReservation.Allocations { + allocatedVMIDs[vmUUID] = struct{}{} + } + } + + candidates := filterPaygCandidates(tc.hv.Name, tc.hv, tc.vms, fg, allocatedVMIDs) + + if tc.wantVMIDs != nil { + if len(candidates) != len(tc.wantVMIDs) { + t.Fatalf("want %d candidates, got %d: %v", len(tc.wantVMIDs), len(candidates), candidates) + } + for i, want := range tc.wantVMIDs { + if candidates[i].VMID != want { + t.Errorf("candidates[%d]: want VMID %q, got %q", i, want, candidates[i].VMID) + } + } + } else if len(candidates) != tc.wantCount { + t.Errorf("want %d candidates, got %d", tc.wantCount, len(candidates)) + } + }) + } +} + +// ============================================================================ +// Tests: ScanAZForPaygCandidates +// ============================================================================ + +func TestScanAZForPaygCandidates(t *testing.T) { + fg := testFlavorGroup() + + tests := []struct { + name string + hvs []*hv1.Hypervisor + vmSourceVMs []reservations.VM + vmSourceErr error + az string + projectID string + wantHVs []string // HV names expected in result (non-empty candidate lists) + wantErr bool + }{ + { + name: "HV in AZ with matching PAYG VM — returned", + hvs: []*hv1.Hypervisor{hvWithInstances("host-1", "az-1", activeInstance("vm-a"))}, + vmSourceVMs: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + az: "az-1", + projectID: "project-1", + wantHVs: []string{"host-1"}, + }, + { + name: "HV in different AZ — not returned", + hvs: []*hv1.Hypervisor{hvWithInstances("host-1", "az-2", activeInstance("vm-a"))}, + vmSourceVMs: []reservations.VM{vmOnHV("vm-a", "host-1", "small", 8192)}, + az: "az-1", + projectID: "project-1", + wantHVs: nil, + }, + { + name: "VMSource returns error — propagated", + hvs: []*hv1.Hypervisor{hvWithInstances("host-1", "az-1", activeInstance("vm-a"))}, + vmSourceErr: errors.New("db error"), + az: "az-1", + projectID: "project-1", + wantErr: true, + }, + { + name: "no HVs in AZ — returns nil, no error", + hvs: []*hv1.Hypervisor{}, + az: "az-1", + projectID: "project-1", + wantHVs: nil, + }, + { + name: "VMSource returns no VMs — returns empty, no error", + hvs: []*hv1.Hypervisor{hvWithInstances("host-1", "az-1", activeInstance("vm-a"))}, + vmSourceVMs: nil, + az: "az-1", + projectID: "project-1", + wantHVs: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + objs := make([]client.Object, len(tc.hvs)) + for i, hv := range tc.hvs { + objs[i] = hv + } + k8sClient := buildPaygClient(t, objs...) + vmSource := &fakeVMSource{vms: tc.vmSourceVMs, err: tc.vmSourceErr} + + result, err := ScanAZForPaygCandidates(context.Background(), k8sClient, vmSource, tc.az, tc.projectID, fg) + if tc.wantErr { + if err == nil { + t.Fatal("expected error, got nil") + } + return + } + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + + for _, hvName := range tc.wantHVs { + if _, ok := result[hvName]; !ok { + t.Errorf("expected candidates for HV %q, not present in result", hvName) + } + } + // No unexpected HVs. + for hvName := range result { + found := false + for _, want := range tc.wantHVs { + if want == hvName { + found = true + break + } + } + if !found { + t.Errorf("unexpected HV %q in result", hvName) + } + } + }) + } +} diff --git a/internal/scheduling/reservations/commitments/payg_rollback_test.go b/internal/scheduling/reservations/commitments/payg_rollback_test.go new file mode 100644 index 000000000..c9962436a --- /dev/null +++ b/internal/scheduling/reservations/commitments/payg_rollback_test.go @@ -0,0 +1,118 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package commitments + +import ( + "context" + "testing" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/api/resource" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +// TestCRLifecycle_PAYGRollback verifies that a pre-allocated PAYG reservation slot survives +// rollback after a failed resize attempt. +// +// Scenario: +// 1. CR accepted at 4 GiB via PAYG pre-allocation (no scheduler call needed). +// 2. CR resized to 8 GiB; 2nd slot (blind scheduler path) is rejected. +// 3. Controller rolls back to AcceptedSpec: original pre-allocated slot preserved intact. +func TestCRLifecycle_PAYGRollback(t *testing.T) { + objects := []client.Object{ + newTestFlavorKnowledge(), + intgHypervisorWithAZ("host-1", "test-az", "vm-payg-rollback"), + } + env := newIntgEnv(t, objects, intgRejectScheduler, &fakeVMSource{vms: []reservations.VM{{ + UUID: "vm-payg-rollback", + FlavorName: "test-flavor", + CurrentHypervisor: "host-1", + }}}) + defer env.close() + + cr := intgCRAllowRejection("my-cr", "uuid-payg-rollback-0001", v1alpha1.CommitmentStatusConfirmed) + if err := env.k8sClient.Create(context.Background(), cr); err != nil { + t.Fatalf("create CR: %v", err) + } + + // Phase 1: PAYG pre-allocation → Accepted (scheduler rejects but isn't called for PAYG slot). + intgDriveToTerminal(t, env, []string{cr.Name}) + crState := env.getCR(t, cr.Name) + if !meta.IsStatusConditionTrue(crState.Status.Conditions, v1alpha1.CommittedResourceConditionReady) { + t.Fatalf("phase 1: expected CR to be Ready=True after PAYG pre-allocation") + } + + slots := env.listChildReservations(t, cr.Name) + if len(slots) != 1 { + t.Fatalf("phase 1: want 1 pre-allocated slot, got %d", len(slots)) + } + if slots[0].Spec.TargetHost != "host-1" { + t.Errorf("phase 1: expected pre-allocated slot on host-1, got %q", slots[0].Spec.TargetHost) + } + if _, ok := slots[0].Spec.CommittedResourceReservation.Allocations["vm-payg-rollback"]; !ok { + t.Error("phase 1: expected vm-payg-rollback in Spec.Allocations") + } + preAllocatedSlotName := slots[0].Name + + // Phase 2: resize to 8 GiB (2 slots required). PAYG VM already allocated → 2nd slot goes to + // blind scheduler path → rejected. + patch := client.MergeFrom(crState.DeepCopy()) + crState.Spec.Amount = resource.MustParse("8Gi") + if err := env.k8sClient.Patch(context.Background(), &crState, patch); err != nil { + t.Fatalf("patch CR to 8Gi: %v", err) + } + + ctx := context.Background() + crReq := ctrl.Request{NamespacedName: types.NamespacedName{Name: cr.Name}} + + if _, err := env.crController.Reconcile(ctx, crReq); err != nil { + t.Fatalf("phase 2 CR reconcile: %v", err) + } + var resList v1alpha1.ReservationList + if err := env.k8sClient.List(ctx, &resList, client.MatchingLabels{ + v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource, + }); err != nil { + t.Fatalf("list reservations: %v", err) + } + for _, res := range resList.Items { + resReq := ctrl.Request{NamespacedName: types.NamespacedName{Name: res.Name}} + if _, err := env.resController.Reconcile(ctx, resReq); err != nil { + t.Fatalf("reservation reconcile %s (pass 1): %v", res.Name, err) + } + if _, err := env.resController.Reconcile(ctx, resReq); err != nil { + t.Fatalf("reservation reconcile %s (pass 2): %v", res.Name, err) + } + } + // CR controller: 2nd slot Ready=False → rollbackToAccepted → Rejected. + if _, err := env.crController.Reconcile(ctx, crReq); err != nil { + t.Fatalf("phase 2 final CR reconcile: %v", err) + } + + // Rollback must preserve exactly the original pre-allocated slot. + var finalList v1alpha1.ReservationList + if err := env.k8sClient.List(ctx, &finalList, client.MatchingLabels{ + v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource, + }); err != nil { + t.Fatalf("list reservations after rollback: %v", err) + } + if len(finalList.Items) != 1 { + t.Fatalf("rollback: want 1 slot, got %d", len(finalList.Items)) + } + surviving := finalList.Items[0] + if surviving.Name != preAllocatedSlotName { + t.Errorf("rollback: expected original pre-allocated slot %q to survive, got %q", preAllocatedSlotName, surviving.Name) + } + if surviving.Spec.TargetHost != "host-1" { + t.Errorf("rollback: expected TargetHost host-1, got %q", surviving.Spec.TargetHost) + } + if _, ok := surviving.Spec.CommittedResourceReservation.Allocations["vm-payg-rollback"]; !ok { + t.Error("rollback: expected vm-payg-rollback in Spec.Allocations of surviving slot") + } + intgAssertCRCondition(t, env.k8sClient, []string{cr.Name}, metav1.ConditionFalse, v1alpha1.CommittedResourceReasonRejected) +} diff --git a/internal/scheduling/reservations/commitments/reservation_controller.go b/internal/scheduling/reservations/commitments/reservation_controller.go index db24c0ed5..598f3a667 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller.go +++ b/internal/scheduling/reservations/commitments/reservation_controller.go @@ -26,9 +26,13 @@ import ( "github.com/cobaltcore-dev/cortex/api/scheduling" "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + "github.com/cobaltcore-dev/cortex/pkg/keystone" "github.com/cobaltcore-dev/cortex/pkg/multicluster" + "github.com/cobaltcore-dev/cortex/pkg/sso" 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 @@ -41,6 +45,10 @@ type CommitmentReservationController struct { Conf ReservationControllerConfig // SchedulerClient for making scheduler API calls. SchedulerClient *reservations.SchedulerClient + // DomainResolver resolves OpenStack domain IDs to domain names so the + // 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 } // echoParentGeneration copies Spec.CommittedResourceReservation.ParentGeneration to @@ -213,6 +221,25 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr availabilityZone = res.Spec.AvailabilityZone } + // Resolve domain name for the domain_name scheduler hint consumed by + // filter_external_customer to enforce host restrictions for external customers. + // Resolution is best-effort: if the DomainResolver is not configured (no + // KeystoneSecretRef) or the lookup fails, we log and proceed without the hint. + // filter_external_customer already handles a missing hint by skipping the filter, + // so omitting it degrades gracefully rather than blocking scheduling. + schedulerHints := map[string]any{ + "_nova_check_type": string(schedulerdelegationapi.ReserveForCommittedResourceIntent), + } + if r.DomainResolver != nil && res.Spec.CommittedResourceReservation != nil && res.Spec.CommittedResourceReservation.DomainID != "" { + domainName, err := r.DomainResolver.ResolveDomainName(ctx, res.Spec.CommittedResourceReservation.DomainID) + if err != nil { + logger.Error(err, "failed to resolve domain name for scheduler hint, proceeding without it", + "domainID", res.Spec.CommittedResourceReservation.DomainID) + } else { + schedulerHints["domain_name"] = domainName + } + } + // Get flavor details from flavor group knowledge CRD knowledge := &reservations.FlavorGroupKnowledgeClient{Client: r.Client} flavorGroups, err := knowledge.GetAllFlavorGroups(ctx, nil) @@ -281,11 +308,7 @@ func (r *CommitmentReservationController) Reconcile(ctx context.Context, req ctr EligibleHosts: eligibleHosts, Pipeline: pipelineName, AvailabilityZone: availabilityZone, - // Set hint to indicate this is a CR reservation scheduling request. - // This prevents other CR reservations from being unlocked during capacity filtering. - SchedulerHints: map[string]any{ - "_nova_check_type": string(schedulerdelegationapi.ReserveForCommittedResourceIntent), - }, + SchedulerHints: schedulerHints, } scheduleOpts := scheduling.Options{ ReadOnly: false, // mutates state (reservation placement) @@ -422,9 +445,12 @@ func (r *CommitmentReservationController) reconcileAllocations(ctx context.Conte allocationAge := now.Sub(allocation.CreationTimestamp.Time) isInGracePeriod := allocationAge < r.Conf.AllocationGracePeriod.Duration - if isInGracePeriod { - // New allocation: VM may not yet appear in the HV CRD (still spawning). - // Signal to requeue with the short grace-period interval; skip verification. + // Confirmed VMs (already in Status.Allocations) bypass the grace period: + // their departure from the HV CRD is authoritative and must be acted on immediately. + // Unconfirmed VMs still within the grace period may not yet appear in the HV CRD + // (still spawning), so defer verification and requeue with a short interval. + _, isConfirmed := existingStatusAllocations[vmUUID] + if !isConfirmed && isInGracePeriod { result.HasAllocationsInGracePeriod = true logger.V(1).Info("allocation in grace period, deferring verification", "vm", vmUUID, @@ -432,7 +458,7 @@ func (r *CommitmentReservationController) reconcileAllocations(ctx context.Conte continue } - // Post-grace-period: use HV CRD as authoritative source. + // Post-grace-period or confirmed VM: use HV CRD as authoritative source. if hvInstanceSet[vmUUID] { newStatusAllocations[vmUUID] = expectedHost logger.V(1).Info("verified VM allocation via Hypervisor CRD", @@ -599,6 +625,39 @@ func (r *CommitmentReservationController) hypervisorToReservations(ctx context.C func (r *CommitmentReservationController) Init(ctx context.Context, conf ReservationControllerConfig) error { r.SchedulerClient = reservations.NewSchedulerClient(conf.SchedulerURL) logf.FromContext(ctx).Info("scheduler client initialized for commitment reservation controller", "url", conf.SchedulerURL) + + if conf.KeystoneSecretRef.Name != "" { + var authenticatedHTTP *http.Client + if conf.SSOSecretRef != nil { + var err error + authenticatedHTTP, err = sso.Connector{Client: r.Client}.FromSecretRef(ctx, *conf.SSOSecretRef) + if err != nil { + return fmt.Errorf("failed to initialize SSO client for domain resolver: %w", err) + } + } + keystoneClient, err := keystone.Connector{Client: r.Client, HTTPClient: authenticatedHTTP}.FromSecretRef(ctx, conf.KeystoneSecretRef) + if err != nil { + return fmt.Errorf("failed to initialize keystone client for domain resolver: %w", err) + } + provider := keystoneClient.Client() + identityURL, err := provider.EndpointLocator(gophercloud.EndpointOpts{ + Type: "identity", + Availability: gophercloud.Availability(keystoneClient.Availability()), + }) + if err != nil { + return fmt.Errorf("failed to locate keystone identity endpoint for domain resolver: %w", err) + } + sc := &gophercloud.ServiceClient{ + ProviderClient: provider, + Endpoint: identityURL, + Type: "identity", + } + r.DomainResolver = newKeystoneDomainResolver(sc) + logf.FromContext(ctx).Info("domain resolver initialized for commitment reservation controller") + } else { + logf.FromContext(ctx).Info("keystoneSecretRef not configured — domain_name scheduler hint will not be set for CR reservations") + } + return nil } diff --git a/internal/scheduling/reservations/commitments/reservation_controller_test.go b/internal/scheduling/reservations/commitments/reservation_controller_test.go index 8c5817893..651852c2c 100644 --- a/internal/scheduling/reservations/commitments/reservation_controller_test.go +++ b/internal/scheduling/reservations/commitments/reservation_controller_test.go @@ -6,6 +6,7 @@ package commitments import ( "context" "encoding/json" + "errors" "net/http" "net/http/httptest" "testing" @@ -21,8 +22,27 @@ import ( schedulerdelegationapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" ) +// mockDomainResolver is a test double for DomainResolver. +type mockDomainResolver struct { + // names maps domainID → name returned on success. + names map[string]string + // err is returned for all calls when non-nil. + err error + // calls records every domainID passed to ResolveDomainName. + calls []string +} + +func (m *mockDomainResolver) ResolveDomainName(_ context.Context, domainID string) (string, error) { + m.calls = append(m.calls, domainID) + if m.err != nil { + return "", m.err + } + return m.names[domainID], nil +} + func TestCommitmentReservationController_Reconcile(t *testing.T) { scheme := newCRTestScheme(t) @@ -305,6 +325,82 @@ func TestReconcileAllocations_HypervisorCRDPath(t *testing.T) { } } +// TestReconcileAllocations_ConfirmedVMDeparture verifies that a VM already confirmed in +// Status.Allocations is removed immediately when it disappears from the HV CRD, without +// waiting for the grace period to expire. +func TestReconcileAllocations_ConfirmedVMDeparture(t *testing.T) { + scheme := newCRTestScheme(t) + config := ReservationControllerConfig{AllocationGracePeriod: metav1.Duration{Duration: 15 * time.Minute}} + now := time.Now() + + // VM was written to Spec.Allocations very recently (within grace period) but was already + // confirmed (present in Status.Allocations). It has since disappeared from the HV CRD. + recentTime := metav1.NewTime(now.Add(-2 * time.Minute)) + + res := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "test-reservation"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + TargetHost: "host-1", + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + ProjectID: "test-project", + ResourceName: "test-flavor", + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + "vm-confirmed": { + CreationTimestamp: recentTime, + Resources: map[hv1.ResourceName]resource.Quantity{ + "memory": resource.MustParse("4Gi"), + }, + }, + }, + }, + }, + Status: v1alpha1.ReservationStatus{ + Host: "host-1", + Conditions: []metav1.Condition{ + {Type: v1alpha1.ReservationConditionReady, Status: metav1.ConditionTrue, Reason: "ReservationActive"}, + }, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationStatus{ + // VM was previously confirmed — it is in Status.Allocations. + Allocations: map[string]string{"vm-confirmed": "host-1"}, + }, + }, + } + + // HV CRD exists but the VM is gone — simulates termination/evacuation. + hv := newTestHypervisorCRD("host-1", []hv1.Instance{}) + + k8sClient := newCRTestClient(scheme, res, hv) + controller := &CommitmentReservationController{Client: k8sClient, Scheme: scheme, Conf: config} + + ctx := WithNewGlobalRequestID(context.Background()) + result, err := controller.reconcileAllocations(ctx, res) + if err != nil { + t.Fatalf("reconcileAllocations() error = %v", err) + } + + // Must not be treated as grace-period — departure of confirmed VM is immediate. + if result.HasAllocationsInGracePeriod { + t.Error("confirmed VM departure must not trigger grace period requeue") + } + + var updated v1alpha1.Reservation + if err := k8sClient.Get(ctx, client.ObjectKeyFromObject(res), &updated); err != nil { + t.Fatalf("get updated reservation: %v", err) + } + + // Spec.Allocations must be empty — VM removed. + if n := len(updated.Spec.CommittedResourceReservation.Allocations); n != 0 { + t.Errorf("expected 0 spec allocations after departure, got %d", n) + } + // Status.Allocations must also be empty — updated in the same reconcile pass. + if updated.Status.CommittedResourceReservation == nil || + len(updated.Status.CommittedResourceReservation.Allocations) != 0 { + t.Errorf("expected 0 status allocations after departure, got %#v", + updated.Status.CommittedResourceReservation) + } +} + // newTestCRReservation creates a test CR reservation with allocations on "host-1". func newTestCRReservation(allocations map[string]metav1.Time) *v1alpha1.Reservation { const host = "host-1" @@ -770,3 +866,119 @@ func TestCommitmentReservationController_reconcileInstanceReservation_Success(t t.Errorf("Expected host %v, got %v", "test-host-1", updated.Status.Host) } } + +// ============================================================================ +// Test: domain_name scheduler hint +// ============================================================================ + +// newTestSchedulerServer captures the decoded request and returns a single host. +func newTestSchedulerServer(t *testing.T, captured *schedulerdelegationapi.ExternalSchedulerRequest) *httptest.Server { + t.Helper() + return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if err := json.NewDecoder(r.Body).Decode(captured); err != nil { + t.Errorf("failed to decode scheduler request: %v", err) + w.WriteHeader(http.StatusBadRequest) + return + } + resp := schedulerdelegationapi.ExternalSchedulerResponse{Hosts: []string{"test-host-1"}} + if err := json.NewEncoder(w).Encode(resp); err != nil { + t.Fatalf("failed to write scheduler response: %v", err) + } + })) +} + +func TestCommitmentReservationController_DomainNameHint(t *testing.T) { + tests := []struct { + name string + resolver DomainResolver + wantDomainName string // empty means hint must be absent + wantResolverCall string // expected domainID passed to resolver; empty means no call expected + }{ + { + name: "hint present when resolver succeeds", + resolver: &mockDomainResolver{names: map[string]string{"domain-uuid-1": "monsoon3"}}, + wantDomainName: "monsoon3", + wantResolverCall: "domain-uuid-1", + }, + { + name: "hint absent when resolver fails, scheduling continues", + resolver: &mockDomainResolver{err: errors.New("keystone unavailable")}, + wantDomainName: "", + }, + { + name: "hint absent when resolver is nil (keystoneSecretRef not configured)", + resolver: nil, + wantDomainName: "", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + scheme := newCRTestScheme(t) + reservation := &v1alpha1.Reservation{ + ObjectMeta: ctrl.ObjectMeta{Name: "test-reservation"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + ProjectID: "test-project", + DomainID: "domain-uuid-1", + ResourceName: "test-flavor", + }, + Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: resource.MustParse("4Gi"), + hv1.ResourceCPU: resource.MustParse("2"), + }, + }, + } + k8sClient := newCRTestClient(scheme, reservation, newTestFlavorKnowledge(), + &hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: "test-host-1"}}) + + var captured schedulerdelegationapi.ExternalSchedulerRequest + server := newTestSchedulerServer(t, &captured) + defer server.Close() + + reconciler := &CommitmentReservationController{ + Client: k8sClient, + Scheme: scheme, + Conf: ReservationControllerConfig{SchedulerURL: server.URL}, + SchedulerClient: reservations.NewSchedulerClient(server.URL), + DomainResolver: tt.resolver, + } + + if _, err := reconciler.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: reservation.Name}, + }); err != nil { + t.Fatalf("Reconcile() error = %v", err) + } + + // Verify resolver was called correctly. + if tt.wantResolverCall != "" { + m := tt.resolver.(*mockDomainResolver) + if len(m.calls) != 1 || m.calls[0] != tt.wantResolverCall { + t.Errorf("expected resolver called with %q, got %v", tt.wantResolverCall, m.calls) + } + } + + // Verify domain_name hint. + gotDomain, err := captured.Spec.Data.GetSchedulerHintStr("domain_name") + if tt.wantDomainName == "" { + if err == nil { + t.Errorf("expected domain_name hint absent, got %q", gotDomain) + } + } else { + if err != nil { + t.Fatalf("expected domain_name hint %q, got error: %v", tt.wantDomainName, err) + } + if gotDomain != tt.wantDomainName { + t.Errorf("expected domain_name hint %q, got %q", tt.wantDomainName, gotDomain) + } + } + + // _nova_check_type must always be present regardless of resolver outcome. + gotIntent, err := captured.Spec.Data.GetSchedulerHintStr("_nova_check_type") + if err != nil || gotIntent != string(schedulerdelegationapi.ReserveForCommittedResourceIntent) { + t.Errorf("expected _nova_check_type=%q, got %q (err=%v)", schedulerdelegationapi.ReserveForCommittedResourceIntent, gotIntent, err) + } + }) + } +} diff --git a/internal/scheduling/reservations/commitments/reservation_manager.go b/internal/scheduling/reservations/commitments/reservation_manager.go index 75c3cda7f..ddd509b19 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager.go +++ b/internal/scheduling/reservations/commitments/reservation_manager.go @@ -10,6 +10,7 @@ import ( "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins/compute" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "github.com/go-logr/logr" apierrors "k8s.io/apimachinery/pkg/api/errors" @@ -47,21 +48,30 @@ func (e *SlotLimitExceededError) Error() string { return fmt.Sprintf("commitment would create %d new reservation slots, exceeds limit of %d", e.NewSlots, e.Limit) } -// ReservationManager handles CRUD operations for Reservation CRDs. -type ReservationManager struct { - client.Client +// ReservationManagerConfig holds options for ReservationManager. +type ReservationManagerConfig struct { // SlotCreationDelay adds a pause between consecutive Reservation CRD creates to spread // scheduler load across time rather than bursting all creates at once. SlotCreationDelay time.Duration // MaxSlots caps the total number of Reservation CRDs for a single commitment. // When non-zero, ApplyCommitmentState returns an error if the desired slot count would // exceed this limit. Only set by the caller on the AllowRejection=true (API) path. - MaxSlots int + MaxSlots int + EnablePaygPreAllocation bool + VMSource reservations.VMSource +} + +// ReservationManager handles CRUD operations for Reservation CRDs. +type ReservationManager struct { + client.Client + cfg ReservationManagerConfig } -func NewReservationManager(k8sClient client.Client) *ReservationManager { +// NewReservationManager creates a ReservationManager using the given client and config. +func NewReservationManager(k8sClient client.Client, cfg ReservationManagerConfig) *ReservationManager { return &ReservationManager{ Client: k8sClient, + cfg: cfg, } } @@ -92,7 +102,6 @@ func (m *ReservationManager) ApplyCommitmentState( log = log.WithName("reservation-manager") - // Phase 1: List and filter existing reservations for this commitment var allReservations v1alpha1.ReservationList if err := m.List(ctx, &allReservations, client.MatchingLabels{ v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource, @@ -100,7 +109,6 @@ func (m *ReservationManager) ApplyCommitmentState( return nil, fmt.Errorf("failed to list reservations: %w", err) } - // Filter by CommitmentUUID to find reservations for this commitment var existing []v1alpha1.Reservation for _, res := range allReservations.Items { if res.Spec.CommittedResourceReservation != nil && @@ -109,7 +117,6 @@ func (m *ReservationManager) ApplyCommitmentState( } } - // Phase 2: Calculate memory delta (desired - current) flavorGroup, exists := flavorGroups[desiredState.FlavorGroupName] if !exists { @@ -124,7 +131,6 @@ func (m *ReservationManager) ApplyCommitmentState( deltaMemoryBytes -= memoryQuantity.Value() } - // Log only if there's actual work to do (delta != 0) hasChanges := deltaMemoryBytes != 0 nextSlotIndex := GetNextSlotIndex(existing) @@ -194,16 +200,86 @@ func (m *ReservationManager) ApplyCommitmentState( } } - // Phase 5 (CREATE): Create new reservations (capacity increased) + // Phase 4.5 (PAYG PRE-ALLOCATE): absorb existing PAYG VMs into pre-populated slots. + // Creates one slot per PAYG VM (largest-first), consuming the delta. + // Any remaining delta falls through to Phase 5 (blind scheduler). + if m.cfg.EnablePaygPreAllocation && m.cfg.VMSource != nil && deltaMemoryBytes > 0 && desiredState.AvailabilityZone != "" { + scanStart := time.Now() + candidatesByHV, scanErr := ScanAZForPaygCandidates( + ctx, m.Client, m.cfg.VMSource, + desiredState.AvailabilityZone, desiredState.ProjectID, flavorGroup, + ) + if scanErr != nil { + log.Error(scanErr, "PAYG candidate scan failed, falling back to blind scheduling") + } else { + var allCandidates []PAYGCandidate + for _, hvCandidates := range candidatesByHV { + allCandidates = append(allCandidates, hvCandidates...) + } + sortCandidatesDesc(allCandidates) + + for deltaMemoryBytes > 0 && len(allCandidates) > 0 { + // Largest candidate that fits the delta (candidates sorted descending). + // If none fit, pick the smallest to minimise waste on the undersized slot. + idx := len(allCandidates) - 1 + for i, c := range allCandidates { + if int64(c.MemoryMB)*1024*1024 <= deltaMemoryBytes { //nolint:gosec // bounded by flavor specs + idx = i + break + } + } + candidate := allCandidates[idx] + allCandidates = append(allCandidates[:idx], allCandidates[idx+1:]...) + + candidateMemBytes := int64(candidate.MemoryMB) * 1024 * 1024 //nolint:gosec // bounded by flavor specs + slotMemoryBytes := candidateMemBytes + if slotMemoryBytes > deltaMemoryBytes { + slotMemoryBytes = deltaMemoryBytes + } + slotMemoryMB := uint64(slotMemoryBytes) / (1024 * 1024) + reservation := m.newPaygReservation(desiredState, nextSlotIndex, candidate, slotMemoryMB, flavorGroup, creator) + deltaMemoryBytes -= slotMemoryBytes + result.Created++ + result.TouchedReservations = append(result.TouchedReservations, *reservation) + + if err := m.Create(ctx, reservation); err != nil { + if apierrors.IsAlreadyExists(err) { + return result, fmt.Errorf("reservation %s already exists (collision detected): %w", reservation.Name, err) + } + return result, fmt.Errorf("failed to create PAYG reservation slot %d: %w", nextSlotIndex, err) + } + log.Info("created PAYG pre-allocated reservation slot", + "slot", reservation.Name, + "host", candidate.HVName, + "vm", candidate.VMID, + "flavorName", candidate.FlavorName, + "slotMemoryMB", slotMemoryMB, + ) + nextSlotIndex++ + + if m.cfg.SlotCreationDelay > 0 && deltaMemoryBytes > 0 { + timer := time.NewTimer(m.cfg.SlotCreationDelay) + select { + case <-ctx.Done(): + timer.Stop() + return result, ctx.Err() + case <-timer.C: + } + } + } + } + log.Info("PAYG remapping done", "slotsCreated", result.Created, "durationMs", time.Since(scanStart).Milliseconds()) + } + if deltaMemoryBytes > 0 { newSlots := countNewSlots(deltaMemoryBytes, flavorGroup) - if m.MaxSlots > 0 && newSlots > m.MaxSlots { - return nil, &SlotLimitExceededError{NewSlots: newSlots, Limit: m.MaxSlots} + if m.cfg.MaxSlots > 0 && newSlots > m.cfg.MaxSlots { + return nil, &SlotLimitExceededError{NewSlots: newSlots, Limit: m.cfg.MaxSlots} } log.Info("creating reservation slots", "commitmentUUID", desiredState.CommitmentUUID, "slots", newSlots, - "slotCreationDelay", m.SlotCreationDelay, + "slotCreationDelay", m.cfg.SlotCreationDelay, ) } for deltaMemoryBytes > 0 { @@ -229,8 +305,8 @@ func (m *ReservationManager) ApplyCommitmentState( // Throttle: pause between consecutive creates to spread scheduler load. // Skip after the last slot (deltaMemoryBytes <= 0) — no follow-up create to defer. - if m.SlotCreationDelay > 0 && deltaMemoryBytes > 0 { - timer := time.NewTimer(m.SlotCreationDelay) + if m.cfg.SlotCreationDelay > 0 && deltaMemoryBytes > 0 { + timer := time.NewTimer(m.cfg.SlotCreationDelay) select { case <-ctx.Done(): timer.Stop() @@ -252,7 +328,6 @@ func (m *ReservationManager) ApplyCommitmentState( } } - // Only log if there were actual changes if hasChanges || result.Created > 0 || len(result.RemovedReservations) > 0 || result.Repaired > 0 { log.Info("commitment state sync completed", "commitmentUUID", desiredState.CommitmentUUID, @@ -274,9 +349,6 @@ func (m *ReservationManager) syncReservationMetadata( state *CommitmentState, ) (*v1alpha1.Reservation, error) { - // if any of CommitmentUUID, DomainID, StartTime, EndTime, ParentGeneration differ from desired state, need to patch. - // AvailabilityZone is intentionally excluded: an AZ mismatch is handled in Phase 3 (delete + recreate) - // because the reservation is pinned to a host and cannot simply be patched to a different AZ. if (state.CommitmentUUID != "" && reservation.Spec.CommittedResourceReservation.CommitmentUUID != state.CommitmentUUID) || (state.DomainID != "" && reservation.Spec.CommittedResourceReservation.DomainID != state.DomainID) || (state.StartTime != nil && (reservation.Spec.StartTime == nil || !reservation.Spec.StartTime.Time.Equal(*state.StartTime))) || @@ -311,9 +383,8 @@ func (m *ReservationManager) syncReservationMetadata( } return reservation, nil - } else { - return nil, nil // No changes needed } + return nil, nil // No changes needed } // selectFlavor picks the largest flavor whose memory fits within deltaMemoryBytes. @@ -359,8 +430,6 @@ func (m *ReservationManager) newReservation( } name := fmt.Sprintf("%s%d", namePrefix, slotIndex) - // Select largest flavor that fits remaining memory (flavors sorted descending by memory then vCPUs). - // This works for both fixed and varying CPU:RAM ratio groups. flavorInGroup, memoryBytes := selectFlavor(deltaMemoryBytes, flavorGroup) cpus := int64(flavorInGroup.VCPUs) //nolint:gosec // VCPUs from flavor specs, realistically bounded @@ -389,12 +458,89 @@ func (m *ReservationManager) newReservation( }, } - // Set AvailabilityZone if specified if state.AvailabilityZone != "" { spec.AvailabilityZone = state.AvailabilityZone } - // Set validity times if specified + if state.StartTime != nil { + spec.StartTime = &metav1.Time{Time: *state.StartTime} + } + if state.EndTime != nil { + spec.EndTime = &metav1.Time{Time: *state.EndTime} + } + + return &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{ + v1alpha1.LabelReservationType: v1alpha1.ReservationTypeLabelCommittedResource, + }, + Annotations: map[string]string{ + v1alpha1.AnnotationCreatorRequestID: state.CreatorRequestID, + }, + }, + Spec: spec, + } +} + +// newPaygReservation creates a Reservation pre-populated with a PAYG VM. +// The slot uses the candidate's exact flavor name and the given slotMemoryMB (which may be +// less than candidate.MemoryMB when the remaining CR delta is smaller than the VM's memory). +// VCPUs are taken from the flavor group to keep the slot spec consistent. +func (m *ReservationManager) newPaygReservation( + state *CommitmentState, + slotIndex int, + candidate PAYGCandidate, + slotMemoryMB uint64, + flavorGroup compute.FlavorGroupFeature, + creator string, +) *v1alpha1.Reservation { + + namePrefix := state.NamePrefix + if namePrefix == "" { + namePrefix = fmt.Sprintf("commitment-%s-", state.CommitmentUUID) + } + name := fmt.Sprintf("%s%d", namePrefix, slotIndex) + + var cpus int64 + for _, f := range flavorGroup.Flavors { + if f.Name == candidate.FlavorName { + cpus = int64(f.VCPUs) //nolint:gosec // VCPUs from flavor specs, realistically bounded + break + } + } + + memoryBytes := int64(slotMemoryMB) * 1024 * 1024 //nolint:gosec // bounded by CR amount + spec := v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + SchedulingDomain: v1alpha1.SchedulingDomainNova, + TargetHost: candidate.HVName, + Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: *resource.NewQuantity(memoryBytes, resource.BinarySI), + hv1.ResourceCPU: *resource.NewQuantity(cpus, resource.DecimalSI), + }, + CommittedResourceReservation: &v1alpha1.CommittedResourceReservationSpec{ + ProjectID: state.ProjectID, + CommitmentUUID: state.CommitmentUUID, + DomainID: state.DomainID, + ResourceGroup: state.FlavorGroupName, + ResourceName: candidate.FlavorName, + Creator: creator, + ParentGeneration: state.ParentGeneration, + Allocations: map[string]v1alpha1.CommittedResourceAllocation{ + candidate.VMID: { + CreationTimestamp: metav1.Now(), + Resources: map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceMemory: *resource.NewQuantity(memoryBytes, resource.BinarySI), + hv1.ResourceCPU: *resource.NewQuantity(cpus, resource.DecimalSI), + }, + }, + }, + }, + } + if state.AvailabilityZone != "" { + spec.AvailabilityZone = state.AvailabilityZone + } if state.StartTime != nil { spec.StartTime = &metav1.Time{Time: *state.StartTime} } diff --git a/internal/scheduling/reservations/commitments/reservation_manager_test.go b/internal/scheduling/reservations/commitments/reservation_manager_test.go index d890bd496..2346a16f2 100644 --- a/internal/scheduling/reservations/commitments/reservation_manager_test.go +++ b/internal/scheduling/reservations/commitments/reservation_manager_test.go @@ -9,6 +9,7 @@ import ( "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins/compute" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "github.com/go-logr/logr" "k8s.io/apimachinery/pkg/api/resource" @@ -368,8 +369,7 @@ func TestApplyCommitmentState(t *testing.T) { objects[i] = &tt.existingSlots[i] } k8sClient := newCRTestClient(scheme, objects...) - manager := NewReservationManager(k8sClient) - manager.MaxSlots = tt.maxSlots + manager := NewReservationManager(k8sClient, ReservationManagerConfig{MaxSlots: tt.maxSlots}) flavorGroups := testFlavorGroups() if tt.flavorGroupOverride != nil { @@ -430,6 +430,314 @@ func TestApplyCommitmentState(t *testing.T) { } } +// ============================================================================ +// Tests: ApplyCommitmentState — PAYG pre-allocation (Phase 4.5) +// ============================================================================ + +func TestApplyCommitmentState_PAYG(t *testing.T) { + const ( + az = "test-az" + projectID = "project-1" + hvName = "host-1" + vmUUID = "vm-payg-1" + ) + fg := testFlavorGroup() // small=8GiB, medium=16GiB, large=32GiB + + // hvWithAZ returns an HV in az with one active instance. + hvWithAZ := func(name string, instanceIDs ...string) *hv1.Hypervisor { + instances := make([]hv1.Instance, len(instanceIDs)) + for i, id := range instanceIDs { + instances[i] = hv1.Instance{ID: id, Name: id, Active: true} + } + return &hv1.Hypervisor{ + ObjectMeta: metav1.ObjectMeta{ + Name: name, + Labels: map[string]string{"topology.kubernetes.io/zone": az}, + }, + Status: hv1.HypervisorStatus{Instances: instances}, + } + } + + // paygVM returns a VM matching the test flavor group on hvName. + paygVM := func(uuid, flavorName string) reservations.VM { + return reservations.VM{ + UUID: uuid, + FlavorName: flavorName, + CurrentHypervisor: hvName, + } + } + + tests := []struct { + name string + hypervisors []*hv1.Hypervisor + paygVMs []reservations.VM // returned by fake VMSource + existingSlots []v1alpha1.Reservation + desiredMemoryGiB int64 + enablePayg bool // must be set true to activate PAYG; false = flag disabled (Phase 5 only) + validateTouched func(t *testing.T, touched []v1alpha1.Reservation) + }{ + { + name: "PAYG VM found — slot created pre-allocated with TargetHost and Allocations", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, vmUUID)}, + paygVMs: []reservations.VM{paygVM(vmUUID, "small")}, + enablePayg: true, + desiredMemoryGiB: 8, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + res := touched[0] + if res.Spec.TargetHost != hvName { + t.Errorf("TargetHost: want %q, got %q", hvName, res.Spec.TargetHost) + } + if res.Spec.CommittedResourceReservation == nil { + t.Fatal("CommittedResourceReservation is nil") + } + if _, ok := res.Spec.CommittedResourceReservation.Allocations[vmUUID]; !ok { + t.Errorf("expected vm %q in Spec.Allocations", vmUUID) + } + if res.Spec.CommittedResourceReservation.ResourceName != "small" { + t.Errorf("ResourceName: want %q, got %q", "small", res.Spec.CommittedResourceReservation.ResourceName) + } + }, + }, + { + name: "no PAYG VMs — falls through to Phase 5 (no TargetHost set)", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName)}, // HV has no instances + paygVMs: nil, + enablePayg: true, + desiredMemoryGiB: 8, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot from Phase 5, got %d", len(touched)) + } + if touched[0].Spec.TargetHost != "" { + t.Errorf("expected no TargetHost for blind slot, got %q", touched[0].Spec.TargetHost) + } + if len(touched[0].Spec.CommittedResourceReservation.Allocations) != 0 { + t.Error("expected no pre-allocations for blind slot") + } + }, + }, + { + name: "PAYG VM already allocated — excluded, slot goes to Phase 5", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, vmUUID)}, + paygVMs: []reservations.VM{paygVM(vmUUID, "small")}, + enablePayg: true, + existingSlots: []v1alpha1.Reservation{ + // Different commitment UUID so Phase 1 doesn't count it against our delta. + func() v1alpha1.Reservation { + s := withAZ(newTestCRSlot("other-cr-0", 8, hvName, "test-group", + map[string]v1alpha1.CommittedResourceAllocation{vmUUID: {}}), az) + s.Spec.CommittedResourceReservation.CommitmentUUID = "other-uuid" + return s + }(), + }, + desiredMemoryGiB: 8, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + if touched[0].Spec.TargetHost != "" { + t.Errorf("expected Phase 5 blind slot (no TargetHost), got %q", touched[0].Spec.TargetHost) + } + }, + }, + { + name: "PAYG VM larger than remaining delta — slot undersized, VM pre-allocated", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, vmUUID)}, + paygVMs: []reservations.VM{paygVM(vmUUID, "small")}, + enablePayg: true, + // delta = 4 GiB < VM 8 GiB — undersize path + desiredMemoryGiB: 4, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + res := touched[0] + if res.Spec.TargetHost != hvName { + t.Errorf("TargetHost: want %q, got %q", hvName, res.Spec.TargetHost) + } + if _, ok := res.Spec.CommittedResourceReservation.Allocations[vmUUID]; !ok { + t.Errorf("expected vm %q in Spec.Allocations", vmUUID) + } + wantMem := int64(4) * 1024 * 1024 * 1024 + memQty := res.Spec.Resources[hv1.ResourceMemory] + gotMem := memQty.Value() + if gotMem != wantMem { + t.Errorf("slot memory: want %d, got %d", wantMem, gotMem) + } + }, + }, + { + name: "PAYG covers part of delta — remaining goes to Phase 5", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, vmUUID)}, + paygVMs: []reservations.VM{paygVM(vmUUID, "small")}, + enablePayg: true, + // delta = 16 GiB; PAYG covers 8, remaining 8 → Phase 5 + desiredMemoryGiB: 16, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 2 { + t.Fatalf("want 2 slots, got %d", len(touched)) + } + var preAllocated, blind int + for _, res := range touched { + if res.Spec.TargetHost != "" { + preAllocated++ + } else { + blind++ + } + } + if preAllocated != 1 { + t.Errorf("want 1 pre-allocated slot, got %d", preAllocated) + } + if blind != 1 { + t.Errorf("want 1 blind slot, got %d", blind) + } + }, + }, + { + name: "EnablePaygPreAllocation=false — PAYG VM ignored, slot goes to Phase 5", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, vmUUID)}, + paygVMs: []reservations.VM{paygVM(vmUUID, "small")}, + enablePayg: false, + desiredMemoryGiB: 8, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + if touched[0].Spec.TargetHost != "" { + t.Errorf("expected Phase 5 blind slot (no TargetHost), got %q", touched[0].Spec.TargetHost) + } + if len(touched[0].Spec.CommittedResourceReservation.Allocations) != 0 { + t.Error("expected no pre-allocations when flag is disabled") + } + }, + }, + { + // delta=8GiB, candidates=[large=32GiB, small=8GiB]: best-fit picks small (exact fit), + // not large (which would produce an undersized 8GiB slot on a 32GiB VM). + name: "best-fit: fitting candidate preferred over larger one", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, "vm-large", "vm-small")}, + paygVMs: []reservations.VM{paygVM("vm-large", "large"), paygVM("vm-small", "small")}, + enablePayg: true, + desiredMemoryGiB: 8, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + if _, ok := touched[0].Spec.CommittedResourceReservation.Allocations["vm-small"]; !ok { + t.Errorf("expected vm-small (8GiB exact fit), got allocations: %v", + touched[0].Spec.CommittedResourceReservation.Allocations) + } + }, + }, + { + // delta=4GiB, candidates=[large=32GiB, small=8GiB]: no exact fit — picks smallest + // oversized (small=8GiB) to minimise waste on the undersized slot. + name: "undersize fallback: smallest oversized candidate chosen", + hypervisors: []*hv1.Hypervisor{hvWithAZ(hvName, "vm-large", "vm-small")}, + paygVMs: []reservations.VM{paygVM("vm-large", "large"), paygVM("vm-small", "small")}, + enablePayg: true, + desiredMemoryGiB: 4, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + if len(touched) != 1 { + t.Fatalf("want 1 slot, got %d", len(touched)) + } + if _, ok := touched[0].Spec.CommittedResourceReservation.Allocations["vm-small"]; !ok { + t.Errorf("expected smallest oversized vm-small, got allocations: %v", + touched[0].Spec.CommittedResourceReservation.Allocations) + } + wantMem := int64(4) * 1024 * 1024 * 1024 + memQty := touched[0].Spec.Resources[hv1.ResourceMemory] + if got := memQty.Value(); got != wantMem { + t.Errorf("slot memory: want %d (delta), got %d", wantMem, got) + } + }, + }, + { + // delta=40GiB, candidates=[large=32GiB, medium=16GiB, small=8GiB]: + // round 1: largest fit = large (32GiB), delta remaining = 8GiB + // round 2: largest fit = small (8GiB), delta remaining = 0 + // → 2 pre-allocated slots, large and small used; medium untouched. + name: "multi-round: best-fit applied each round from remaining candidates", + hypervisors: []*hv1.Hypervisor{ + hvWithAZ(hvName, "vm-large", "vm-medium", "vm-small"), + }, + paygVMs: []reservations.VM{ + paygVM("vm-large", "large"), + paygVM("vm-medium", "medium"), + paygVM("vm-small", "small"), + }, + enablePayg: true, + desiredMemoryGiB: 40, + validateTouched: func(t *testing.T, touched []v1alpha1.Reservation) { + // 2 PAYG slots + 1 blind slot for remaining (40-32-8=0, so actually just 2) + var preAllocated int + usedVMs := make(map[string]bool) + for _, res := range touched { + if res.Spec.TargetHost != "" { + preAllocated++ + for vm := range res.Spec.CommittedResourceReservation.Allocations { + usedVMs[vm] = true + } + } + } + if preAllocated != 2 { + t.Errorf("want 2 pre-allocated slots, got %d", preAllocated) + } + if !usedVMs["vm-large"] { + t.Error("expected vm-large (32GiB) to be used in round 1") + } + if !usedVMs["vm-small"] { + t.Error("expected vm-small (8GiB) to be used in round 2") + } + if usedVMs["vm-medium"] { + t.Error("expected vm-medium (16GiB) to be skipped (large+small = exact fit)") + } + }, + }, + } + + scheme := newCRTestScheme(t) + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + objects := make([]client.Object, 0, len(tt.hypervisors)+len(tt.existingSlots)) + for _, hv := range tt.hypervisors { + objects = append(objects, hv) + } + for i := range tt.existingSlots { + objects = append(objects, &tt.existingSlots[i]) + } + k8sClient := newCRTestClient(scheme, objects...) + + mgr := NewReservationManager(k8sClient, ReservationManagerConfig{ + EnablePaygPreAllocation: tt.enablePayg, + VMSource: &fakeVMSource{vms: tt.paygVMs}, + }) + + desiredState := &CommitmentState{ + CommitmentUUID: "abc123", + ProjectID: projectID, + FlavorGroupName: "test-group", + TotalMemoryBytes: tt.desiredMemoryGiB * 1024 * 1024 * 1024, + AvailabilityZone: az, + } + + result, err := mgr.ApplyCommitmentState( + context.Background(), logr.Discard(), desiredState, map[string]compute.FlavorGroupFeature{"test-group": fg}, "test", + ) + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + if tt.validateTouched != nil { + tt.validateTouched(t, result.TouchedReservations) + } + }) + } +} + // ============================================================================ // Tests: newReservation flavor selection // ============================================================================ diff --git a/internal/scheduling/reservations/commitments/usage_reconciler.go b/internal/scheduling/reservations/commitments/usage_reconciler.go index 18c53eee6..b15aa0777 100644 --- a/internal/scheduling/reservations/commitments/usage_reconciler.go +++ b/internal/scheduling/reservations/commitments/usage_reconciler.go @@ -57,7 +57,7 @@ func (r *UsageReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl cr.Status.UsedResources = nil cr.Status.LastUsageReconcileAt = nil cr.Status.UsageObservedGeneration = nil - cr.Status.StatusSummary = v1alpha1.ComputeStatusSummary(cr.Spec, cr.Status, start) + cr.Status.StatusSummary = computeStatusSummary(cr.Spec, cr.Status, start) if err := r.Status().Patch(ctx, &cr, client.MergeFrom(old)); err != nil { return ctrl.Result{}, client.IgnoreNotFound(err) } @@ -76,7 +76,7 @@ func (r *UsageReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl cr.Status.UsedResources = nil cr.Status.LastUsageReconcileAt = nil cr.Status.UsageObservedGeneration = nil - cr.Status.StatusSummary = v1alpha1.ComputeStatusSummary(cr.Spec, cr.Status, start) + cr.Status.StatusSummary = computeStatusSummary(cr.Spec, cr.Status, start) if err := r.Status().Patch(ctx, &cr, client.MergeFrom(old)); err != nil { return ctrl.Result{}, client.IgnoreNotFound(err) } @@ -224,7 +224,7 @@ func (r *UsageReconciler) writeUsageStatus(ctx context.Context, state *Commitmen } target.Status.LastUsageReconcileAt = &now target.Status.UsageObservedGeneration = &target.Generation - target.Status.StatusSummary = v1alpha1.ComputeStatusSummary(target.Spec, target.Status, now.Time) + target.Status.StatusSummary = computeStatusSummary(target.Spec, target.Status, now.Time) return r.Status().Patch(ctx, target, client.MergeFrom(old)) } diff --git a/internal/scheduling/reservations/failover/controller.go b/internal/scheduling/reservations/failover/controller.go index 04dd8d401..73e471f1d 100644 --- a/internal/scheduling/reservations/failover/controller.go +++ b/internal/scheduling/reservations/failover/controller.go @@ -268,6 +268,28 @@ func (c *FailoverReservationController) ReconcilePeriodic(ctx context.Context) ( allHypervisors = append(allHypervisors, hv.Name) } + // Build a set of VM UUIDs currently active on any known hypervisor. + // This is used as a safeguard against removing failover allocations when the + // VM source (postgres) is missing data but the VM is still alive on a + // hypervisor (e.g. after a postgres data loss / restore). Without this + // cross-check a wiped postgres would cause every failover reservation to be + // emptied and then deleted. + vmsOnHypervisor := make(map[string]string) + for _, hv := range hypervisorList.Items { + for _, inst := range hv.Status.Instances { + if !inst.Active { + continue + } + if _, exists := vmsOnHypervisor[inst.ID]; exists { + // VM appears on multiple hypervisors (transient during live + // migration). Keep the first occurrence; the safeguard only + // needs to know it exists somewhere. + continue + } + vmsOnHypervisor[inst.ID] = hv.Name + } + } + // 2. Get all VMs that might need failover reservations vms, err := c.VMSource.ListVMsOnHypervisors(ctx, &hypervisorList, c.Config.TrustHypervisorLocation) if err != nil { @@ -289,7 +311,7 @@ func (c *FailoverReservationController) ReconcilePeriodic(ctx context.Context) ( logger.V(1).Info("found failover reservations", "count", len(failoverReservations)) // 3. Remove VMs from reservations if they are no longer valid - failoverReservations, reservationsToUpdate := reconcileRemoveInvalidVMFromReservations(ctx, vms, failoverReservations) + failoverReservations, reservationsToUpdate := reconcileRemoveInvalidVMFromReservations(ctx, vms, vmsOnHypervisor, failoverReservations) for _, res := range reservationsToUpdate { if err := c.patchReservationStatus(ctx, res); err != nil { @@ -381,9 +403,17 @@ func (c *FailoverReservationController) ReconcilePeriodic(ctx context.Context) ( // - The VM has moved to a different host // Returns the updated list of reservations (with modifications applied in-memory). // The caller is responsible for persisting any changes to the cluster. +// +// vmsOnHypervisor maps VM UUID -> hypervisor name for every VM currently active +// on any known hypervisor (sourced from hv1.HypervisorList.Status.Instances). +// It is used as a safeguard: if a VM is missing from the postgres-derived vms +// slice but is still present on a hypervisor, we keep the allocation. This +// protects failover reservations from being wiped by transient/total postgres +// data loss. func reconcileRemoveInvalidVMFromReservations( ctx context.Context, vms []reservations.VM, + vmsOnHypervisor map[string]string, failoverReservations []v1alpha1.Reservation, ) (updatedReservations []v1alpha1.Reservation, reservationsToUpdate []*v1alpha1.Reservation) { @@ -404,6 +434,18 @@ func reconcileRemoveInvalidVMFromReservations( for vmUUID, allocatedHypervisor := range allocations { vmCurrentHypervisor, vmExists := vmToHypervisor[vmUUID] if !vmExists { + // Safeguard: if the VM is missing from the VM source (e.g. + // postgres) but is still reported active on a hypervisor by + // the hypervisor operator CRD, keep the allocation. This + // prevents a postgres data loss from cascading into a mass + // deletion of failover reservations. + if hv, stillOnHV := vmsOnHypervisor[vmUUID]; stillOnHV { + logger.Info("keeping VM allocation despite missing from VM source: still active on hypervisor", + "vmUUID", vmUUID, "reservation", res.Name, + "allocatedHypervisor", allocatedHypervisor, "hypervisor", hv) + updatedAllocations[vmUUID] = allocatedHypervisor + continue + } logger.Info("removing VM from reservation allocations because VM no longer exists", "vmUUID", vmUUID, "reservation", res.Name) needsUpdate = true diff --git a/internal/scheduling/reservations/failover/controller_test.go b/internal/scheduling/reservations/failover/controller_test.go index 482fa4f30..55a09cc03 100644 --- a/internal/scheduling/reservations/failover/controller_test.go +++ b/internal/scheduling/reservations/failover/controller_test.go @@ -417,6 +417,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { tests := []struct { name string vms []reservations.VM + vmsOnHypervisor map[string]string reservations []v1alpha1.Reservation expectedUpdatedCount int // number of reservations in updatedReservations expectedToUpdateCount int // number of reservations that need cluster update @@ -444,7 +445,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { name: "VM no longer exists - remove from allocations", vms: []reservations.VM{ newTestVM("vm-1", "host1"), - // vm-2 no longer exists + // vm-2 no longer exists (and not on any hypervisor) }, reservations: []v1alpha1.Reservation{ newTestReservation("res-1", "host3", map[string]string{ @@ -481,7 +482,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { vms: []reservations.VM{ newTestVM("vm-1", "host1"), newTestVM("vm-2", "host2"), - // vm-3 no longer exists + // vm-3 no longer exists (and not on any hypervisor) }, reservations: []v1alpha1.Reservation{ newTestReservation("res-1", "host3", map[string]string{ @@ -502,7 +503,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { { name: "all VMs removed from reservation - empty allocations", vms: []reservations.VM{ - // no VMs exist + // no VMs exist (and not on any hypervisor) }, reservations: []v1alpha1.Reservation{ newTestReservation("res-1", "host3", map[string]string{ @@ -545,7 +546,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { vms: []reservations.VM{ newTestVM("vm-1", "host1"), // valid newTestVM("vm-2", "host5"), // moved from host2 to host5 - // vm-3 deleted + // vm-3 deleted (and not on any hypervisor) newTestVM("vm-4", "host4"), // valid }, reservations: []v1alpha1.Reservation{ @@ -565,6 +566,105 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { "res-2": {"vm-4": "host4"}, }, }, + // ==================================================================== + // Safeguard: VM missing from VM source but still on a hypervisor + // (e.g. postgres data loss). Allocations must be preserved. + // ==================================================================== + { + name: "safeguard: VM missing from VM source but still on hypervisor - keep allocation", + vms: []reservations.VM{ + newTestVM("vm-1", "host1"), + // vm-2 missing from VM source (postgres lost data) + }, + vmsOnHypervisor: map[string]string{ + "vm-1": "host1", + "vm-2": "host2", // still alive on hypervisor + }, + reservations: []v1alpha1.Reservation{ + newTestReservation("res-1", "host3", map[string]string{ + "vm-1": "host1", + "vm-2": "host2", + }), + }, + expectedUpdatedCount: 1, + expectedToUpdateCount: 0, // safeguard prevents the update + expectedAllocationsPerRes: map[string]map[string]string{ + "res-1": {"vm-1": "host1", "vm-2": "host2"}, + }, + }, + { + name: "safeguard: VM source completely empty but VMs still on hypervisors - keep all allocations", + vms: []reservations.VM{ + // VM source returns nothing (postgres wiped) + }, + vmsOnHypervisor: map[string]string{ + "vm-1": "host1", + "vm-2": "host2", + "vm-3": "host3", + }, + reservations: []v1alpha1.Reservation{ + newTestReservation("res-1", "host4", map[string]string{ + "vm-1": "host1", + "vm-2": "host2", + }), + newTestReservation("res-2", "host5", map[string]string{ + "vm-3": "host3", + }), + }, + expectedUpdatedCount: 2, + expectedToUpdateCount: 0, // safeguard prevents both updates + expectedAllocationsPerRes: map[string]map[string]string{ + "res-1": {"vm-1": "host1", "vm-2": "host2"}, + "res-2": {"vm-3": "host3"}, + }, + }, + { + name: "safeguard: only some missing VMs are still on hypervisor - remove only fully gone ones", + vms: []reservations.VM{ + newTestVM("vm-1", "host1"), + // vm-2 missing from postgres but on hypervisor + // vm-3 missing from both + }, + vmsOnHypervisor: map[string]string{ + "vm-1": "host1", + "vm-2": "host2", + // vm-3 not on any hypervisor + }, + reservations: []v1alpha1.Reservation{ + newTestReservation("res-1", "host4", map[string]string{ + "vm-1": "host1", + "vm-2": "host2", // safeguarded + "vm-3": "host3", // truly gone - remove + }), + }, + expectedUpdatedCount: 1, + expectedToUpdateCount: 1, + expectedAllocationsPerRes: map[string]map[string]string{ + "res-1": {"vm-1": "host1", "vm-2": "host2"}, + }, + }, + { + name: "safeguard does not protect against hypervisor mismatch (VM is in postgres on different host)", + vms: []reservations.VM{ + newTestVM("vm-1", "host1"), + newTestVM("vm-2", "host5"), // moved to host5 in postgres + }, + vmsOnHypervisor: map[string]string{ + "vm-1": "host1", + "vm-2": "host5", + }, + reservations: []v1alpha1.Reservation{ + newTestReservation("res-1", "host3", map[string]string{ + "vm-1": "host1", + "vm-2": "host2", // host mismatch - should still be removed + }), + }, + expectedUpdatedCount: 1, + expectedToUpdateCount: 1, + expectedAllocationsPerRes: map[string]map[string]string{ + "res-1": {"vm-1": "host1"}, + }, + }, } for _, tt := range tests { @@ -573,6 +673,7 @@ func TestReconcileRemoveInvalidVMFromReservations(t *testing.T) { updatedReservations, reservationsToUpdate := reconcileRemoveInvalidVMFromReservations( ctx, tt.vms, + tt.vmsOnHypervisor, tt.reservations, )