From 34aa2f4c3c8f45779281a6e879393eff908ddf1b Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Thu, 18 Jun 2026 11:50:15 +0200 Subject: [PATCH 1/9] Implement in-flight reservations controller --- .../reservations/inflight/controller.go | 276 ++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 internal/scheduling/reservations/inflight/controller.go diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go new file mode 100644 index 000000000..a31e072c0 --- /dev/null +++ b/internal/scheduling/reservations/inflight/controller.go @@ -0,0 +1,276 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package inflight + +import ( + "context" + "errors" + "time" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/pkg/multicluster" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + apierrors "k8s.io/apimachinery/pkg/api/errors" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/predicate" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +var ( + idxReservationByTargetHost = "spec.targetHost" + idxReservationByTargetHostFn = func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok { + return nil + } + if res.Spec.TargetHost == "" { + return nil + } + return []string{res.Spec.TargetHost} + } +) + +// Controller owns the lifecycle of in-flight reservations. +type Controller struct{ client.Client } + +// Reconcile is part of the main kubernetes reconciliation loop which aims to +// move the current state of the cluster closer to the desired state. +// +// For more details, check Reconcile and its Result here: +// - https://pkg.go.dev/sigs.k8s.io/controller-runtime@v0.20.2/pkg/reconcile +// +// For more details about the method shape, read up here: +// - https://ahmet.im/blog/controller-pitfalls/#reconcile-method-shape +func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { + log := ctrl.LoggerFrom(ctx) + log.V(1).Info("Reconciling resource") + + obj := new(v1alpha1.Reservation) + if err := c.Get(ctx, req.NamespacedName, obj); err != nil { + if apierrors.IsNotFound(err) { + // If the custom resource is not found then it usually means + // that it was deleted or not created. + log.Info("Resource not found. Ignoring since object must be deleted") + return ctrl.Result{}, nil + } + // Error reading the object - requeue the request. + log.Error(err, "Failed to get resource") + return ctrl.Result{}, err + } + + // Sanity checks which should always succeed due to the predicate. + if obj.Spec.Type != v1alpha1.ReservationTypeInFlight { + log.Error(errors.New("unexpected reservation type"), + "Received a reservation with an unexpected type", + "reservationType", obj.Spec.Type) + orig := obj.DeepCopy() + meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionFalse, + Reason: "UnexpectedType", + Message: "Expected reservation type to be InFlightReservation", + }) + return ctrl.Result{}, c.Status().Patch(ctx, obj, client.MergeFrom(orig)) + } + if obj.Spec.InFlightReservation == nil { + log.Error(errors.New("missing in-flight reservation spec"), + "Received a reservation with missing in-flight reservation spec") + orig := obj.DeepCopy() + meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionFalse, + Reason: "MissingSpec", + Message: "In-flight reservation spec is required when type is InFlightReservation", + }) + return ctrl.Result{}, c.Status().Patch(ctx, obj, client.MergeFrom(orig)) + } + + // Get a list of all hypervisors and check if the instance + // has spawned on any of them. + hvs := new(hv1.HypervisorList) + if err := c.List(ctx, hvs); err != nil { + log.Error(err, "Failed to list hypervisors") + return ctrl.Result{}, err + } + found := false + hypervisorName := "" + for _, hv := range hvs.Items { + for _, instance := range hv.Status.Instances { + if instance.ID == obj.Spec.InFlightReservation.VMID { + found = true + hypervisorName = hv.Name + break + } + } + if found { + break + } + } + + if !found { + // The instance has not spawned on any hypervisor (yet). + // Requeue and check again later. + log.V(1).Info("Instance has not spawned on any hypervisor yet, requeuing", + "vmID", obj.Spec.InFlightReservation.VMID) + // TODO: monitor this & alert + meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionUnknown, + Reason: "InstanceNotFound", + Message: "The instance has not spawned on any hypervisor yet", + }) + if err := c.Status().Update(ctx, obj); err != nil { + log.Error(err, "Failed to update reservation status") + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + + // We don't care where the instance came up. Even if this in-flight + // reservation is for another host, we can prune it. + log.Info("Instance has spawned on a hypervisor, removing in-flight reservation", + "vmID", obj.Spec.InFlightReservation.VMID, + "hypervisor", hypervisorName) + return ctrl.Result{}, c.Delete(ctx, obj) +} + +// handleReservations generates a new event handler for in flight reservations. +func (c *Controller) handleReservations() handler.EventHandler { + handler := handler.Funcs{} + handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ + Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd + }}) + } + handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ + Name: evt.ObjectOld.(*v1alpha1.Reservation).Name, // cluster-scoped crd + }}) + } + handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ + Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd + }}) + } + return handler +} + +// predicateReservations generates a new predicate for in flight reservations. +func (c *Controller) predicateReservations() predicate.Predicate { + return predicate.NewPredicateFuncs(func(object client.Object) bool { + reservation, ok := object.(*v1alpha1.Reservation) + if !ok { + return false // Not a Reservation object. + } + if reservation.Spec.Type != v1alpha1.ReservationTypeInFlight { + return false // Not an in-flight reservation. + } + return true // Reconcile. + }) +} + +// handleHypervisors generates a new event handler for hypervisors. +func (c *Controller) handleHypervisors() handler.EventHandler { + handler := handler.Funcs{} + enqueueCorrespondingReservations := func(ctx context.Context, hvName string, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + log := ctrl.LoggerFrom(ctx) + log.V(1).Info("Enqueuing reservations corresponding to hypervisor", + "hypervisor", hvName) + // Requeue all reservations targeting this hypervisor, since the + // instance list has changed and we might find the instance for + // some in-flight reservation now. + reservations := &v1alpha1.ReservationList{} + if err := c.List(ctx, reservations, client.MatchingFields{ + idxReservationByTargetHost: hvName, + }); err != nil { + log.Error(err, "Failed to list reservations for hypervisor", + "hypervisor", hvName) + return + } + for _, res := range reservations.Items { + log.V(1).Info("Enqueuing reservation for reconciliation", + "reservation", res.Name, + "targetHost", res.Spec.TargetHost, + "hypervisor", hvName) + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ + Name: res.Name, // cluster-scoped crd + }}) + } + } + handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) + enqueueCorrespondingReservations(ctx, hv.Name, queue) + } + handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.ObjectNew.(*hv1.Hypervisor) + enqueueCorrespondingReservations(ctx, hv.Name, queue) + } + handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, + queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) + enqueueCorrespondingReservations(ctx, hv.Name, queue) + } + return handler +} + +// predicateHypervisors generates a new predicate for hypervisors. +func (c *Controller) predicateHypervisors() predicate.Predicate { + return predicate.NewPredicateFuncs(func(object client.Object) bool { + _, ok := object.(*hv1.Hypervisor) + return ok + }) +} + +// SetupWithManager sets up the controller with the Manager and a multicluster +// client. The multicluster client is used to watch for changes in the +// Reservation CRD across all clusters and trigger reconciliations accordingly. +func (c *Controller) SetupWithManager(ctx context.Context, mgr ctrl.Manager) (err error) { + // Check that the provided client is a multicluster client, since we need + // that to watch for hypervisors across clusters. + mcl, ok := c.Client.(*multicluster.Client) + if !ok { + return errors.New("provided client must be a multicluster client") + } + bldr := multicluster.BuildController(mcl, mgr) + // The reservation crd & hypervisor crd may be distributed across multiple + // remote clusters. + bldr, err = bldr.WatchesMulticluster(&v1alpha1.Reservation{}, + c.handleReservations(), + c.predicateReservations(), + ) + // Index reservations by their target host so we can requeue reservations + // for which the list of instances on a hypervisor has changed. + mcl.IndexField(ctx, + &v1alpha1.Reservation{}, + &v1alpha1.ReservationList{}, + idxReservationByTargetHost, + idxReservationByTargetHostFn, + ) + // Watch hypervisor changes and requeue reservations targeting + // the changed hypervisor. + bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, + c.handleHypervisors(), + c.predicateHypervisors(), + ) + if err != nil { + return err + } + return bldr.Named("inflight-reservation-controller"). + Complete(c) +} From 189de614ce60627e85d4f66830c4152b77c065a2 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Thu, 18 Jun 2026 14:36:41 +0200 Subject: [PATCH 2/9] Tests, polishing, and add to main func --- cmd/manager/main.go | 12 + helm/bundles/cortex-nova/values.yaml | 1 + .../reservations/inflight/controller.go | 16 +- .../reservations/inflight/controller_test.go | 434 ++++++++++++++++++ 4 files changed, 461 insertions(+), 2 deletions(-) create mode 100644 internal/scheduling/reservations/inflight/controller_test.go diff --git a/cmd/manager/main.go b/cmd/manager/main.go index af86e3d84..6da2bf110 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -63,6 +63,7 @@ import ( "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/commitments" commitmentsapi "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/commitments/api" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/failover" + "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/inflight" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations/quota" "github.com/cobaltcore-dev/cortex/pkg/conf" "github.com/cobaltcore-dev/cortex/pkg/monitoring" @@ -475,6 +476,17 @@ func main() { os.Exit(1) } } + if slices.Contains(mainConfig.EnabledControllers, "inflight-reservation-controller") { + setupLog.Info("enabling controller", + "controller", "inflight-reservation-controller") + if err := (&inflight.Controller{ + Client: multiclusterClient, + }).SetupWithManager(ctx, mgr); err != nil { + setupLog.Error(err, "unable to create controller", + "controller", "inflight-reservation-controller") + os.Exit(1) + } + } if slices.Contains(mainConfig.EnabledControllers, "nova-deschedulings-executor") { setupLog.Info("enabling controller", "controller", "nova-deschedulings-executor") executorConfig := conf.GetConfigOrDie[nova.DeschedulingsExecutorConfig]() diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 1171f3b01..53668e306 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -138,6 +138,7 @@ cortex-scheduling-controllers: component: nova-scheduling enabledControllers: - nova-pipeline-controllers + - inflight-reservation-controller - nova-deschedulings-executor - hypervisor-overcommit-controller - committed-resource-reservations-controller diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index a31e072c0..bbd26e974 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -178,6 +178,9 @@ func (c *Controller) predicateReservations() predicate.Predicate { if reservation.Spec.Type != v1alpha1.ReservationTypeInFlight { return false // Not an in-flight reservation. } + if reservation.Spec.SchedulingDomain != v1alpha1.SchedulingDomainNova { + return false // Not a Nova reservation. + } return true // Reconcile. }) } @@ -187,6 +190,7 @@ func (c *Controller) handleHypervisors() handler.EventHandler { handler := handler.Funcs{} enqueueCorrespondingReservations := func(ctx context.Context, hvName string, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + log := ctrl.LoggerFrom(ctx) log.V(1).Info("Enqueuing reservations corresponding to hypervisor", "hypervisor", hvName) @@ -213,16 +217,19 @@ func (c *Controller) handleHypervisors() handler.EventHandler { } handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.ObjectNew.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } @@ -254,14 +261,19 @@ func (c *Controller) SetupWithManager(ctx context.Context, mgr ctrl.Manager) (er c.handleReservations(), c.predicateReservations(), ) + if err != nil { + return err + } // Index reservations by their target host so we can requeue reservations // for which the list of instances on a hypervisor has changed. - mcl.IndexField(ctx, + if err := mcl.IndexField(ctx, &v1alpha1.Reservation{}, &v1alpha1.ReservationList{}, idxReservationByTargetHost, idxReservationByTargetHostFn, - ) + ); err != nil { + return err + } // Watch hypervisor changes and requeue reservations targeting // the changed hypervisor. bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, diff --git a/internal/scheduling/reservations/inflight/controller_test.go b/internal/scheduling/reservations/inflight/controller_test.go new file mode 100644 index 000000000..349e36d8c --- /dev/null +++ b/internal/scheduling/reservations/inflight/controller_test.go @@ -0,0 +1,434 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package inflight + +import ( + "context" + "testing" + "time" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" + "k8s.io/client-go/util/workqueue" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/client/fake" + "sigs.k8s.io/controller-runtime/pkg/event" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// newTestScheme returns a runtime.Scheme with all required types registered. +func newTestScheme(t *testing.T) *runtime.Scheme { + t.Helper() + s := runtime.NewScheme() + if err := v1alpha1.AddToScheme(s); err != nil { + t.Fatalf("failed to add v1alpha1 scheme: %v", err) + } + if err := hv1.AddToScheme(s); err != nil { + t.Fatalf("failed to add hypervisor scheme: %v", err) + } + return s +} + +// newTestClient builds a fake client with the indices the controller relies on. +func newTestClient(scheme *runtime.Scheme, objects ...client.Object) client.Client { + return fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(objects...). + WithStatusSubresource(&v1alpha1.Reservation{}). + WithIndex(&v1alpha1.Reservation{}, idxReservationByTargetHost, func(obj client.Object) []string { + res, ok := obj.(*v1alpha1.Reservation) + if !ok || res.Spec.TargetHost == "" { + return nil + } + return []string{res.Spec.TargetHost} + }). + Build() +} + +// newInFlightReservation builds an in-flight reservation with the given name and VM ID. +func newInFlightReservation(name, vmID, targetHost string) *v1alpha1.Reservation { + return &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: v1alpha1.SchedulingDomainNova, + TargetHost: targetHost, + InFlightReservation: &v1alpha1.InFlightReservationSpec{ + VMID: vmID, + }, + }, + } +} + +// newHypervisor builds a Hypervisor with the given name and instance IDs. +func newHypervisor(name string, instanceIDs ...string) *hv1.Hypervisor { + hv := &hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: name}} + for _, id := range instanceIDs { + hv.Status.Instances = append(hv.Status.Instances, hv1.Instance{ID: id}) + } + return hv +} + +// assertReadyCondition fetches the named reservation and asserts the Ready condition's +// status and reason. Fails fast if the condition is missing. +func assertReadyCondition(t *testing.T, k8sClient client.Client, name string, wantStatus metav1.ConditionStatus, wantReason string) { + t.Helper() + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: name}, &got); err != nil { + t.Fatalf("failed to get reservation %q: %v", name, err) + } + cond := meta.FindStatusCondition(got.Status.Conditions, v1alpha1.ReservationConditionReady) + if cond == nil { + t.Fatalf("Ready condition was not set on %q", name) + return + } + if cond.Status != wantStatus { + t.Errorf("%s: Ready status = %q, want %q", name, cond.Status, wantStatus) + } + if cond.Reason != wantReason { + t.Errorf("%s: Ready reason = %q, want %q", name, cond.Reason, wantReason) + } +} + +func TestReconcile_NotFoundIsIgnored(t *testing.T) { + scheme := newTestScheme(t) + k8sClient := newTestClient(scheme) + c := &Controller{Client: k8sClient} + + res, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "missing"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error for missing object: %v", err) + } + if res.RequeueAfter != 0 { + t.Errorf("expected empty result, got %+v", res) + } +} + +func TestReconcile_UnexpectedTypeSetsConditionFalse(t *testing.T) { + scheme := newTestScheme(t) + wrong := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "wrong-type"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeFailover, + }, + } + k8sClient := newTestClient(scheme, wrong) + c := &Controller{Client: k8sClient} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "wrong-type"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + assertReadyCondition(t, k8sClient, "wrong-type", metav1.ConditionFalse, "UnexpectedType") +} + +func TestReconcile_MissingSpecSetsConditionFalse(t *testing.T) { + scheme := newTestScheme(t) + noSpec := &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: "no-spec"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + // InFlightReservation deliberately nil. + }, + } + k8sClient := newTestClient(scheme, noSpec) + c := &Controller{Client: k8sClient} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "no-spec"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + assertReadyCondition(t, k8sClient, "no-spec", metav1.ConditionFalse, "MissingSpec") +} + +func TestReconcile_InstanceNotSpawnedRequeues(t *testing.T) { + scheme := newTestScheme(t) + res := newInFlightReservation("res-1", "vm-uuid-1", "host-1") + hv := newHypervisor("host-1", "other-vm-uuid") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient} + + result, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 10*time.Second { + t.Errorf("RequeueAfter = %v, want 10s", result.RequeueAfter) + } + + // Reservation still exists. + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("reservation was unexpectedly deleted: %v", err) + } + assertReadyCondition(t, k8sClient, "res-1", metav1.ConditionUnknown, "InstanceNotFound") +} + +func TestReconcile_InstanceSpawnedDeletesReservation(t *testing.T) { + scheme := newTestScheme(t) + res := newInFlightReservation("res-1", "vm-uuid-1", "host-1") + // Instance landed on a *different* host than the target — controller still deletes. + hv1Obj := newHypervisor("host-1") + hv2Obj := newHypervisor("host-2", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv1Obj, hv2Obj) + c := &Controller{Client: k8sClient} + + result, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 0 { + t.Errorf("expected empty result after deletion, got %+v", result) + } + + var got v1alpha1.Reservation + err = k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got) + if err == nil { + t.Fatal("expected reservation to be deleted, but Get succeeded") + } +} + +func TestReconcile_InstanceOnTargetHostDeletesReservation(t *testing.T) { + scheme := newTestScheme(t) + res := newInFlightReservation("res-1", "vm-uuid-1", "host-1") + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err == nil { + t.Fatal("expected reservation to be deleted, but Get succeeded") + } +} + +func TestIdxReservationByTargetHostFn(t *testing.T) { + tests := []struct { + name string + obj client.Object + want []string + }{ + { + name: "wrong type", + obj: &hv1.Hypervisor{}, + want: nil, + }, + { + name: "empty target host", + obj: &v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{TargetHost: ""}, + }, + want: nil, + }, + { + name: "target host set", + obj: &v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{TargetHost: "host-1"}, + }, + want: []string{"host-1"}, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + got := idxReservationByTargetHostFn(tt.obj) + if len(got) != len(tt.want) { + t.Fatalf("got %v, want %v", got, tt.want) + } + for i := range got { + if got[i] != tt.want[i] { + t.Errorf("got[%d] = %q, want %q", i, got[i], tt.want[i]) + } + } + }) + } +} + +func TestPredicateReservations(t *testing.T) { + c := &Controller{} + pred := c.predicateReservations() + + tests := []struct { + name string + obj client.Object + want bool + }{ + { + name: "wrong type", + obj: &hv1.Hypervisor{}, + want: false, + }, + { + name: "wrong reservation type", + obj: &v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeFailover, + SchedulingDomain: v1alpha1.SchedulingDomainNova, + }, + }, + want: false, + }, + { + name: "wrong scheduling domain", + obj: &v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: v1alpha1.SchedulingDomainPods, + }, + }, + want: false, + }, + { + name: "in-flight nova reservation", + obj: &v1alpha1.Reservation{ + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: v1alpha1.SchedulingDomainNova, + }, + }, + want: true, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if got := pred.Create(event.CreateEvent{Object: tt.obj}); got != tt.want { + t.Errorf("Create = %v, want %v", got, tt.want) + } + if got := pred.Update(event.UpdateEvent{ObjectNew: tt.obj, ObjectOld: tt.obj}); got != tt.want { + t.Errorf("Update = %v, want %v", got, tt.want) + } + if got := pred.Delete(event.DeleteEvent{Object: tt.obj}); got != tt.want { + t.Errorf("Delete = %v, want %v", got, tt.want) + } + }) + } +} + +func TestPredicateHypervisors(t *testing.T) { + c := &Controller{} + pred := c.predicateHypervisors() + + if got := pred.Create(event.CreateEvent{Object: &hv1.Hypervisor{}}); !got { + t.Errorf("Create(Hypervisor) = false, want true") + } + if got := pred.Create(event.CreateEvent{Object: &v1alpha1.Reservation{}}); got { + t.Errorf("Create(Reservation) = true, want false") + } +} + +// mockWorkQueue captures items added during handler invocations. +type mockWorkQueue struct { + workqueue.TypedRateLimitingInterface[reconcile.Request] + items []reconcile.Request +} + +func (m *mockWorkQueue) Add(item reconcile.Request) { + m.items = append(m.items, item) +} + +func TestHandleReservations(t *testing.T) { + c := &Controller{} + h := c.handleReservations() + res := &v1alpha1.Reservation{ObjectMeta: metav1.ObjectMeta{Name: "res-1"}} + ctx := context.Background() + + t.Run("Create", func(t *testing.T) { + q := &mockWorkQueue{} + h.Create(ctx, event.CreateEvent{Object: res}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want one entry for res-1", q.items) + } + }) + t.Run("Update", func(t *testing.T) { + q := &mockWorkQueue{} + h.Update(ctx, event.UpdateEvent{ObjectOld: res, ObjectNew: res}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want one entry for res-1", q.items) + } + }) + t.Run("Delete", func(t *testing.T) { + q := &mockWorkQueue{} + h.Delete(ctx, event.DeleteEvent{Object: res}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want one entry for res-1", q.items) + } + }) +} + +func TestHandleHypervisors_EnqueuesMatchingReservations(t *testing.T) { + scheme := newTestScheme(t) + matching := newInFlightReservation("res-1", "vm-1", "host-1") + other := newInFlightReservation("res-2", "vm-2", "host-2") + k8sClient := newTestClient(scheme, matching, other) + c := &Controller{Client: k8sClient} + h := c.handleHypervisors() + + hv := &hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: "host-1"}} + ctx := context.Background() + + t.Run("Create", func(t *testing.T) { + q := &mockWorkQueue{} + h.Create(ctx, event.CreateEvent{Object: hv}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want only res-1", q.items) + } + }) + t.Run("Update", func(t *testing.T) { + q := &mockWorkQueue{} + h.Update(ctx, event.UpdateEvent{ObjectOld: hv, ObjectNew: hv}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want only res-1", q.items) + } + }) + t.Run("Delete", func(t *testing.T) { + q := &mockWorkQueue{} + h.Delete(ctx, event.DeleteEvent{Object: hv}, q) + if len(q.items) != 1 || q.items[0].Name != "res-1" { + t.Errorf("queue = %+v, want only res-1", q.items) + } + }) +} + +func TestHandleHypervisors_NoMatchingReservations(t *testing.T) { + scheme := newTestScheme(t) + other := newInFlightReservation("res-2", "vm-2", "host-2") + k8sClient := newTestClient(scheme, other) + c := &Controller{Client: k8sClient} + h := c.handleHypervisors() + + hv := &hv1.Hypervisor{ObjectMeta: metav1.ObjectMeta{Name: "host-1"}} + q := &mockWorkQueue{} + h.Create(context.Background(), event.CreateEvent{Object: hv}, q) + if len(q.items) != 0 { + t.Errorf("queue = %+v, want empty", q.items) + } +} + +func TestSetupWithManager_RejectsNonMulticlusterClient(t *testing.T) { + scheme := newTestScheme(t) + c := &Controller{Client: newTestClient(scheme)} + err := c.SetupWithManager(context.Background(), nil) + if err == nil { + t.Fatal("expected error for non-multicluster client, got nil") + } +} From 29afa24a4e041c28a9290223ae364e7eb94f3497 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Mon, 6 Jul 2026 13:01:40 +0200 Subject: [PATCH 3/9] Add kpi & alert tracking stalled in-flight reservations --- .../bundles/cortex-nova/templates/alerts.yaml | 37 ++ helm/bundles/cortex-nova/templates/kpis.yaml | 15 + .../plugins/deployment/reservation_state.go | 107 +++++ .../deployment/reservation_state_test.go | 377 ++++++++++++++++++ internal/knowledge/kpis/supported_kpis.go | 11 +- .../reservations/inflight/controller.go | 4 +- 6 files changed, 544 insertions(+), 7 deletions(-) create mode 100644 internal/knowledge/kpis/plugins/deployment/reservation_state.go create mode 100644 internal/knowledge/kpis/plugins/deployment/reservation_state_test.go diff --git a/helm/bundles/cortex-nova/templates/alerts.yaml b/helm/bundles/cortex-nova/templates/alerts.yaml index 840e969f2..836fb1f6a 100644 --- a/helm/bundles/cortex-nova/templates/alerts.yaml +++ b/helm/bundles/cortex-nova/templates/alerts.yaml @@ -303,6 +303,43 @@ spec: configuration. It is recommended to investigate the pipeline status and logs for more details. + - alert: CortexNovaInFlightReservationsBacklogGrowing + # Alert when the number of in-flight reservations stuck with + # reason=InstanceNotFound is trending upwards over a long window, + # rather than staying roughly the same or draining down. A healthy + # scheduler-hypervisor loop should quickly resolve in-flight + # reservations once the VM shows up on any hypervisor; a sustained + # positive derivative indicates the backlog is not being worked off. + # + # deriv() over 1h returns the per-second slope of the gauge; multiplied + # by 3600 it becomes "expected growth per hour". We require this to be + # meaningfully positive AND the current count to be non-trivial, so we + # don't page on a single lingering reservation. + expr: | + deriv(cortex_reservation_state{domain="nova",type="InFlightReservation",state="unknown",reason="InstanceNotFound"}[1h]) * 3600 > 5 + and + cortex_reservation_state{domain="nova",type="InFlightReservation",state="unknown",reason="InstanceNotFound"} > 10 + for: 2h + labels: + context: reservations + dashboard: cortex-status-dashboard/cortex-status-dashboard + service: cortex + severity: warning + support_group: workload-management + playbook: docs/support/playbook/cortex/alerts/unready + annotations: + summary: "In-flight reservation backlog with reason=InstanceNotFound keeps growing" + description: > + The number of in-flight reservations that have not yet observed + their VM on any hypervisor (reason=InstanceNotFound) has been + increasing steadily for at least two hours. Normally these + reservations resolve once the hypervisor operator reports the + instance; a sustained positive trend suggests either VMs are not + landing on hypervisors, or the hypervisor status feed is stalled, + or the in-flight reservation controller is not pruning cleanly. + Investigate the hypervisor operator, the nova build pipeline, and + the in-flight reservation controller logs. + {{- if .Values.kvm.enabled }} - alert: CortexNovaDoesntFindValidKVMHosts expr: sum by (az, hvtype) (increase(cortex_vm_faults{hvtype=~"CH|QEMU",faultmsg=~".*No valid host was found.*",faultmsg!~".*No such host.*"}[5m])) > 0 diff --git a/helm/bundles/cortex-nova/templates/kpis.yaml b/helm/bundles/cortex-nova/templates/kpis.yaml index 5717cd62e..418389e95 100644 --- a/helm/bundles/cortex-nova/templates/kpis.yaml +++ b/helm/bundles/cortex-nova/templates/kpis.yaml @@ -146,6 +146,21 @@ spec: --- apiVersion: cortex.cloud/v1alpha1 kind: KPI +metadata: + name: cortex-nova-reservation-state +spec: + schedulingDomain: nova + impl: reservation_state_kpi + opts: + reservationSchedulingDomain: nova + description: | + This KPI tracks the state of reservation resources managed by cortex, + labelled by reservation type (e.g. InFlightReservation) and the reason + on the Ready condition. It is used to alert on sustained increases of + unready in-flight reservations (e.g. reason=InstanceNotFound). +--- +apiVersion: cortex.cloud/v1alpha1 +kind: KPI metadata: name: vmware-project-utilization spec: diff --git a/internal/knowledge/kpis/plugins/deployment/reservation_state.go b/internal/knowledge/kpis/plugins/deployment/reservation_state.go new file mode 100644 index 000000000..c4608f1f0 --- /dev/null +++ b/internal/knowledge/kpis/plugins/deployment/reservation_state.go @@ -0,0 +1,107 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package deployment + +import ( + "context" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/internal/knowledge/db" + "github.com/cobaltcore-dev/cortex/internal/knowledge/kpis/plugins" + "github.com/cobaltcore-dev/cortex/pkg/conf" + "github.com/prometheus/client_golang/prometheus" + "k8s.io/apimachinery/pkg/api/meta" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +type ReservationStateKPIOpts struct { + // The scheduling domain to filter reservations by. + ReservationSchedulingDomain v1alpha1.SchedulingDomain `json:"reservationSchedulingDomain"` +} + +// KPI observing the state of reservation resources managed by cortex. +// Metrics are labeled by the reservation type (e.g. InFlightReservation, +// CommittedResourceReservation, FailoverReservation) so that operators can +// alert on specific reservation kinds. The state label mirrors the +// v1alpha1.ReservationConditionReady condition status and the reason label +// carries the condition's Reason so that alerts can target specific failure +// modes (e.g. reason="InstanceNotFound" for in-flight reservations whose VM +// has not spawned on any hypervisor yet). +type ReservationStateKPI struct { + // Common base for all KPIs that provides standard functionality. + plugins.BaseKPI[ReservationStateKPIOpts] + + // Prometheus descriptor for the reservation state metric. + counter *prometheus.Desc +} + +func (ReservationStateKPI) GetName() string { return "reservation_state_kpi" } + +// Initialize the KPI. +func (k *ReservationStateKPI) Init(db *db.DB, client client.Client, opts conf.RawOpts) error { + if err := k.BaseKPI.Init(db, client, opts); err != nil { + return err + } + k.counter = prometheus.NewDesc( + "cortex_reservation_state", + "State of cortex managed reservations", + []string{"domain", "type", "state", "reason"}, + nil, + ) + return nil +} + +// Conform to the prometheus collector interface by providing the descriptor. +func (k *ReservationStateKPI) Describe(ch chan<- *prometheus.Desc) { ch <- k.counter } + +// Collect the reservation state metrics. +func (k *ReservationStateKPI) Collect(ch chan<- prometheus.Metric) { + // Get all reservations. The scheduling domain filter is applied per item + // since a Reservation is cluster-scoped and may cover multiple domains. + reservationList := &v1alpha1.ReservationList{} + if err := k.Client.List(context.Background(), reservationList); err != nil { + return + } + // Aggregate counts by (type, state, reason) so that we emit one time + // series per bucket rather than one per reservation. This keeps metric + // cardinality bounded regardless of how many reservations exist. + type bucket struct { + reservationType string + state string + reason string + } + counts := map[bucket]float64{} + for _, r := range reservationList.Items { + if r.Spec.SchedulingDomain != k.Options.ReservationSchedulingDomain { + continue + } + state := "unknown" + reason := "" + cond := meta.FindStatusCondition(r.Status.Conditions, v1alpha1.ReservationConditionReady) + if cond != nil { + reason = cond.Reason + switch cond.Status { + case metav1.ConditionTrue: + state = "ready" + case metav1.ConditionFalse: + state = "error" + default: + state = "unknown" + } + } + counts[bucket{ + reservationType: string(r.Spec.Type), + state: state, + reason: reason, + }]++ + } + for b, v := range counts { + ch <- prometheus.MustNewConstMetric( + k.counter, prometheus.GaugeValue, v, + string(k.Options.ReservationSchedulingDomain), + b.reservationType, b.state, b.reason, + ) + } +} diff --git a/internal/knowledge/kpis/plugins/deployment/reservation_state_test.go b/internal/knowledge/kpis/plugins/deployment/reservation_state_test.go new file mode 100644 index 000000000..19f543ad9 --- /dev/null +++ b/internal/knowledge/kpis/plugins/deployment/reservation_state_test.go @@ -0,0 +1,377 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package deployment + +import ( + "testing" + + "github.com/cobaltcore-dev/cortex/api/v1alpha1" + "github.com/cobaltcore-dev/cortex/pkg/conf" + "github.com/prometheus/client_golang/prometheus" + dto "github.com/prometheus/client_model/go" + v1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "sigs.k8s.io/controller-runtime/pkg/client/fake" +) + +func TestReservationStateKPI_Init(t *testing.T) { + kpi := &ReservationStateKPI{} + if err := kpi.Init(nil, nil, conf.NewRawOpts(`{"reservationSchedulingDomain": "nova"}`)); err != nil { + t.Fatalf("expected no error, got %v", err) + } +} + +func TestReservationStateKPI_GetName(t *testing.T) { + kpi := &ReservationStateKPI{} + expectedName := "reservation_state_kpi" + if name := kpi.GetName(); name != expectedName { + t.Errorf("expected name %q, got %q", expectedName, name) + } +} + +func TestReservationStateKPI_Describe(t *testing.T) { + kpi := &ReservationStateKPI{} + if err := kpi.Init(nil, nil, conf.NewRawOpts(`{"reservationSchedulingDomain": "nova"}`)); err != nil { + t.Fatalf("expected no error, got %v", err) + } + ch := make(chan *prometheus.Desc, 1) + kpi.Describe(ch) + close(ch) + descCount := 0 + for range ch { + descCount++ + } + if descCount != 1 { + t.Errorf("expected 1 descriptor, got %d", descCount) + } +} + +func TestReservationStateKPI_Collect(t *testing.T) { + scheme, err := v1alpha1.SchemeBuilder.Build() + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + + tests := []struct { + name string + reservations []v1alpha1.Reservation + operator v1alpha1.SchedulingDomain + expectedCount int + description string + }{ + { + name: "no reservations", + reservations: []v1alpha1.Reservation{}, + operator: "nova", + expectedCount: 0, + description: "should not collect metrics when no reservations exist", + }, + { + name: "single ready in-flight reservation", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r1"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionTrue, + Reason: "ReservationReady", + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "should collect a single ready metric", + }, + { + name: "unknown in-flight reservation with InstanceNotFound reason", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r2"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionUnknown, + Reason: "InstanceNotFound", + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "should collect a metric labelled with reason=InstanceNotFound", + }, + { + name: "reservation without any conditions falls back to unknown", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r3"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "reservations without conditions should still emit a metric", + }, + { + name: "multiple in-flight reservations with the same reason are aggregated", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r-a"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionUnknown, + Reason: "InstanceNotFound", + }, + }, + }, + }, + { + ObjectMeta: v1.ObjectMeta{Name: "r-b"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionUnknown, + Reason: "InstanceNotFound", + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "two reservations with the same (type,state,reason) should share one time series", + }, + { + name: "different reservation types emit separate metrics", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r-if"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionTrue, + Reason: "ReservationReady", + }, + }, + }, + }, + { + ObjectMeta: v1.ObjectMeta{Name: "r-cr"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeCommittedResource, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionTrue, + Reason: "ReservationReady", + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 2, + description: "in-flight and committed-resource reservations should be labelled separately", + }, + { + name: "filter by scheduling domain", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r-nova"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionTrue, + }, + }, + }, + }, + { + ObjectMeta: v1.ObjectMeta{Name: "r-other"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "ironcore", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionTrue, + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "only reservations matching the configured domain should be counted", + }, + { + name: "error reservation is labelled state=error", + reservations: []v1alpha1.Reservation{ + { + ObjectMeta: v1.ObjectMeta{Name: "r-err"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionFalse, + Reason: "UnexpectedType", + }, + }, + }, + }, + }, + operator: "nova", + expectedCount: 1, + description: "false Ready condition should surface as state=error", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + objects := make([]v1alpha1.Reservation, len(tt.reservations)) + copy(objects, tt.reservations) + + clientBuilder := fake.NewClientBuilder().WithScheme(scheme) + for i := range objects { + clientBuilder = clientBuilder.WithObjects(&objects[i]) + } + client := clientBuilder.Build() + + kpi := &ReservationStateKPI{} + if err := kpi.Init(nil, client, conf.NewRawOpts(`{"reservationSchedulingDomain": "`+string(tt.operator)+`"}`)); err != nil { + t.Fatalf("expected no error, got %v", err) + } + + ch := make(chan prometheus.Metric, 16) + kpi.Collect(ch) + close(ch) + + metricsCount := 0 + for range ch { + metricsCount++ + } + if metricsCount != tt.expectedCount { + t.Errorf("%s: expected %d metrics, got %d", tt.description, tt.expectedCount, metricsCount) + } + }) + } +} + +// TestReservationStateKPI_CollectLabels verifies that the metric emitted for +// an in-flight reservation with reason InstanceNotFound carries the labels +// the InFlightReservationUnready alert relies on. +func TestReservationStateKPI_CollectLabels(t *testing.T) { + scheme, err := v1alpha1.SchemeBuilder.Build() + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + + res := &v1alpha1.Reservation{ + ObjectMeta: v1.ObjectMeta{Name: "r-nf"}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: "nova", + }, + Status: v1alpha1.ReservationStatus{ + Conditions: []v1.Condition{ + { + Type: v1alpha1.ReservationConditionReady, + Status: v1.ConditionUnknown, + Reason: "InstanceNotFound", + }, + }, + }, + } + // Also register a second identical reservation so that we can assert the + // counter carries the aggregate count (2). + res2 := res.DeepCopy() + res2.Name = "r-nf-2" + + client := fake.NewClientBuilder().WithScheme(scheme).WithObjects(res, res2).Build() + + kpi := &ReservationStateKPI{} + if err := kpi.Init(nil, client, conf.NewRawOpts(`{"reservationSchedulingDomain": "nova"}`)); err != nil { + t.Fatalf("expected no error, got %v", err) + } + ch := make(chan prometheus.Metric, 4) + kpi.Collect(ch) + close(ch) + + found := false + for m := range ch { + var metric dto.Metric + if err := m.Write(&metric); err != nil { + t.Fatalf("failed to write metric: %v", err) + } + labels := map[string]string{} + for _, l := range metric.Label { + labels[l.GetName()] = l.GetValue() + } + if labels["type"] != string(v1alpha1.ReservationTypeInFlight) { + continue + } + found = true + if labels["domain"] != "nova" { + t.Errorf("expected domain=nova, got %q", labels["domain"]) + } + if labels["state"] != "unknown" { + t.Errorf("expected state=unknown, got %q", labels["state"]) + } + if labels["reason"] != "InstanceNotFound" { + t.Errorf("expected reason=InstanceNotFound, got %q", labels["reason"]) + } + if got := metric.Gauge.GetValue(); got != 2 { + t.Errorf("expected aggregate count 2, got %f", got) + } + } + if !found { + t.Fatal("expected an in-flight reservation metric to be collected") + } +} diff --git a/internal/knowledge/kpis/supported_kpis.go b/internal/knowledge/kpis/supported_kpis.go index 155e3aab9..d1222e27b 100644 --- a/internal/knowledge/kpis/supported_kpis.go +++ b/internal/knowledge/kpis/supported_kpis.go @@ -29,9 +29,10 @@ var supportedKPIs = map[string]plugins.KPI{ "netapp_storage_pool_cpu_usage_kpi": &storage.NetAppStoragePoolCPUUsageKPI{}, - "datasource_state_kpi": &deployment.DatasourceStateKPI{}, - "knowledge_state_kpi": &deployment.KnowledgeStateKPI{}, - "decision_state_kpi": &deployment.DecisionStateKPI{}, - "kpi_state_kpi": &deployment.KPIStateKPI{}, - "pipeline_state_kpi": &deployment.PipelineStateKPI{}, + "datasource_state_kpi": &deployment.DatasourceStateKPI{}, + "knowledge_state_kpi": &deployment.KnowledgeStateKPI{}, + "decision_state_kpi": &deployment.DecisionStateKPI{}, + "kpi_state_kpi": &deployment.KPIStateKPI{}, + "pipeline_state_kpi": &deployment.PipelineStateKPI{}, + "reservation_state_kpi": &deployment.ReservationStateKPI{}, } diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index bbd26e974..c8a0f64a3 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -116,10 +116,10 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu if !found { // The instance has not spawned on any hypervisor (yet). - // Requeue and check again later. + // Requeue and check again later. We'll alert on this if there are + // too many requeues without the instance spawning. log.V(1).Info("Instance has not spawned on any hypervisor yet, requeuing", "vmID", obj.Spec.InFlightReservation.VMID) - // TODO: monitor this & alert meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ Type: v1alpha1.ReservationConditionReady, Status: metav1.ConditionUnknown, From d869070b432d68cda32c07928c01fc4ad6df7861 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Mon, 6 Jul 2026 13:23:38 +0200 Subject: [PATCH 4/9] Address CodeRabbit review comments - Bound ReservationStateKPI.Collect with a 10s context timeout and log list failures so scrape hangs and errors are visible. - Reuse idxReservationByTargetHostFn in the test fake client so the index logic stays defined in a single place. - Filter Hypervisor Update events in predicateHypervisors so we only re-enqueue reservations when Status.Instances actually changed; unrelated status churn no longer triggers a list + enqueue. - Use Status().Patch(ctx, obj, client.MergeFrom(orig)) in the InstanceNotFound branch to match the other status writes in Reconcile. - Drop unnecessary blank lines after handler function-literal signatures. --- .../plugins/deployment/reservation_state.go | 12 ++++- .../reservations/inflight/controller.go | 46 +++++++++++++------ .../reservations/inflight/controller_test.go | 21 ++++++--- 3 files changed, 58 insertions(+), 21 deletions(-) diff --git a/internal/knowledge/kpis/plugins/deployment/reservation_state.go b/internal/knowledge/kpis/plugins/deployment/reservation_state.go index c4608f1f0..ba6963d53 100644 --- a/internal/knowledge/kpis/plugins/deployment/reservation_state.go +++ b/internal/knowledge/kpis/plugins/deployment/reservation_state.go @@ -5,6 +5,7 @@ package deployment import ( "context" + "time" "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/internal/knowledge/db" @@ -13,9 +14,12 @@ import ( "github.com/prometheus/client_golang/prometheus" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" ) +var reservationStateKPILogger = ctrl.Log.WithName("reservation-state-kpi") + type ReservationStateKPIOpts struct { // The scheduling domain to filter reservations by. ReservationSchedulingDomain v1alpha1.SchedulingDomain `json:"reservationSchedulingDomain"` @@ -58,10 +62,16 @@ func (k *ReservationStateKPI) Describe(ch chan<- *prometheus.Desc) { ch <- k.cou // Collect the reservation state metrics. func (k *ReservationStateKPI) Collect(ch chan<- prometheus.Metric) { + // Bound the list call so a slow API server can't hang the Prometheus + // scrape indefinitely; if it fails we log so the disappearance of the + // reservation metric is not silent. + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() // Get all reservations. The scheduling domain filter is applied per item // since a Reservation is cluster-scoped and may cover multiple domains. reservationList := &v1alpha1.ReservationList{} - if err := k.Client.List(context.Background(), reservationList); err != nil { + if err := k.Client.List(ctx, reservationList); err != nil { + reservationStateKPILogger.Error(err, "Failed to list reservations") return } // Aggregate counts by (type, state, reason) so that we emit one time diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index c8a0f64a3..afc5c7057 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -6,6 +6,7 @@ package inflight import ( "context" "errors" + "reflect" "time" "github.com/cobaltcore-dev/cortex/api/v1alpha1" @@ -120,13 +121,14 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu // too many requeues without the instance spawning. log.V(1).Info("Instance has not spawned on any hypervisor yet, requeuing", "vmID", obj.Spec.InFlightReservation.VMID) + orig := obj.DeepCopy() meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ Type: v1alpha1.ReservationConditionReady, Status: metav1.ConditionUnknown, Reason: "InstanceNotFound", Message: "The instance has not spawned on any hypervisor yet", }) - if err := c.Status().Update(ctx, obj); err != nil { + if err := c.Status().Patch(ctx, obj, client.MergeFrom(orig)); err != nil { log.Error(err, "Failed to update reservation status") return ctrl.Result{}, err } @@ -146,21 +148,18 @@ func (c *Controller) handleReservations() handler.EventHandler { handler := handler.Funcs{} handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) } handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.ObjectOld.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) } handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) @@ -190,7 +189,6 @@ func (c *Controller) handleHypervisors() handler.EventHandler { handler := handler.Funcs{} enqueueCorrespondingReservations := func(ctx context.Context, hvName string, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - log := ctrl.LoggerFrom(ctx) log.V(1).Info("Enqueuing reservations corresponding to hypervisor", "hypervisor", hvName) @@ -217,31 +215,53 @@ func (c *Controller) handleHypervisors() handler.EventHandler { } handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - hv := evt.ObjectNew.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { - hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } return handler } -// predicateHypervisors generates a new predicate for hypervisors. +// predicateHypervisors generates a new predicate for hypervisors. Update +// events are filtered to only trigger when the Status.Instances list actually +// changes, since that is the only field this controller consumes; without +// this filter, unrelated status updates from the hypervisor operator would +// cause a list + enqueue of every reservation targeting the host. func (c *Controller) predicateHypervisors() predicate.Predicate { - return predicate.NewPredicateFuncs(func(object client.Object) bool { - _, ok := object.(*hv1.Hypervisor) - return ok - }) + return predicate.Funcs{ + CreateFunc: func(evt event.CreateEvent) bool { + _, ok := evt.Object.(*hv1.Hypervisor) + return ok + }, + DeleteFunc: func(evt event.DeleteEvent) bool { + _, ok := evt.Object.(*hv1.Hypervisor) + return ok + }, + GenericFunc: func(evt event.GenericEvent) bool { + _, ok := evt.Object.(*hv1.Hypervisor) + return ok + }, + UpdateFunc: func(evt event.UpdateEvent) bool { + oldHV, ok := evt.ObjectOld.(*hv1.Hypervisor) + if !ok { + return false + } + newHV, ok := evt.ObjectNew.(*hv1.Hypervisor) + if !ok { + return false + } + return !reflect.DeepEqual(oldHV.Status.Instances, newHV.Status.Instances) + }, + } } // SetupWithManager sets up the controller with the Manager and a multicluster diff --git a/internal/scheduling/reservations/inflight/controller_test.go b/internal/scheduling/reservations/inflight/controller_test.go index 349e36d8c..9cf6f26d7 100644 --- a/internal/scheduling/reservations/inflight/controller_test.go +++ b/internal/scheduling/reservations/inflight/controller_test.go @@ -41,13 +41,7 @@ func newTestClient(scheme *runtime.Scheme, objects ...client.Object) client.Clie WithScheme(scheme). WithObjects(objects...). WithStatusSubresource(&v1alpha1.Reservation{}). - WithIndex(&v1alpha1.Reservation{}, idxReservationByTargetHost, func(obj client.Object) []string { - res, ok := obj.(*v1alpha1.Reservation) - if !ok || res.Spec.TargetHost == "" { - return nil - } - return []string{res.Spec.TargetHost} - }). + WithIndex(&v1alpha1.Reservation{}, idxReservationByTargetHost, idxReservationByTargetHostFn). Build() } @@ -334,6 +328,19 @@ func TestPredicateHypervisors(t *testing.T) { if got := pred.Create(event.CreateEvent{Object: &v1alpha1.Reservation{}}); got { t.Errorf("Create(Reservation) = true, want false") } + + // Update events must only pass when Status.Instances actually changes, + // so unrelated status churn on the Hypervisor doesn't trigger a list + + // enqueue of all reservations targeting the host. + same := newHypervisor("host-1", "vm-1") + if got := pred.Update(event.UpdateEvent{ObjectOld: same, ObjectNew: same.DeepCopy()}); got { + t.Errorf("Update(no instance change) = true, want false") + } + changed := same.DeepCopy() + changed.Status.Instances = append(changed.Status.Instances, hv1.Instance{ID: "vm-2"}) + if got := pred.Update(event.UpdateEvent{ObjectOld: same, ObjectNew: changed}); !got { + t.Errorf("Update(instance added) = false, want true") + } } // mockWorkQueue captures items added during handler invocations. From c43cfd8650afb312ba50bddc1a722a5cc2913af9 Mon Sep 17 00:00:00 2001 From: "cortex-ai-agents[bot]" <279748396+cortex-ai-agents[bot]@users.noreply.github.com> Date: Mon, 6 Jul 2026 11:45:20 +0000 Subject: [PATCH 5/9] Fix whitespace lint issues in inflight controller Add required blank lines after multi-line function signatures to satisfy the golangci-lint whitespace checker. Co-authored-by: Philipp Matthes --- internal/scheduling/reservations/inflight/controller.go | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index afc5c7057..621be9563 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -148,18 +148,21 @@ func (c *Controller) handleReservations() handler.EventHandler { handler := handler.Funcs{} handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) } handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.ObjectOld.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) } handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + queue.Add(ctrl.Request{NamespacedName: client.ObjectKey{ Name: evt.Object.(*v1alpha1.Reservation).Name, // cluster-scoped crd }}) @@ -189,6 +192,7 @@ func (c *Controller) handleHypervisors() handler.EventHandler { handler := handler.Funcs{} enqueueCorrespondingReservations := func(ctx context.Context, hvName string, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + log := ctrl.LoggerFrom(ctx) log.V(1).Info("Enqueuing reservations corresponding to hypervisor", "hypervisor", hvName) @@ -215,16 +219,19 @@ func (c *Controller) handleHypervisors() handler.EventHandler { } handler.CreateFunc = func(ctx context.Context, evt event.CreateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.UpdateFunc = func(ctx context.Context, evt event.UpdateEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.ObjectNew.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } handler.DeleteFunc = func(ctx context.Context, evt event.DeleteEvent, queue workqueue.TypedRateLimitingInterface[reconcile.Request]) { + hv := evt.Object.(*hv1.Hypervisor) enqueueCorrespondingReservations(ctx, hv.Name, queue) } From efbda50e4d44ff72b61a452dc5d620b82aca464b Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Tue, 7 Jul 2026 15:42:08 +0200 Subject: [PATCH 6/9] Ensure reservations are only pruned if lifecycle op is complete --- cmd/manager/main.go | 9 +- .../reservations/inflight/controller.go | 93 +++++++- .../reservations/inflight/controller_test.go | 220 +++++++++++++++++- .../reservations/inflight/vm_client.go | 96 ++++++++ .../reservations/inflight/vm_client_test.go | 167 +++++++++++++ 5 files changed, 571 insertions(+), 14 deletions(-) create mode 100644 internal/scheduling/reservations/inflight/vm_client.go create mode 100644 internal/scheduling/reservations/inflight/vm_client_test.go diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 6da2bf110..71ecc3e3d 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -479,8 +479,15 @@ func main() { if slices.Contains(mainConfig.EnabledControllers, "inflight-reservation-controller") { setupLog.Info("enabling controller", "controller", "inflight-reservation-controller") + config := conf.GetConfigOrDie[inflight.NovaVMClientConfig]() + novaVMClient, err := inflight.InitNovaVMClient(ctx, multiclusterClient, config) + if err != nil { + setupLog.Error(err, "unable to initialize nova VM client") + os.Exit(1) + } if err := (&inflight.Controller{ - Client: multiclusterClient, + Client: multiclusterClient, + VMClient: novaVMClient, }).SetupWithManager(ctx, mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "inflight-reservation-controller") diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index 621be9563..a19c30eca 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -7,11 +7,14 @@ import ( "context" "errors" "reflect" + "slices" "time" + novaapi "github.com/cobaltcore-dev/cortex/api/external/nova" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/pkg/multicluster" - hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" apierrors "k8s.io/apimachinery/pkg/api/errors" "k8s.io/apimachinery/pkg/api/meta" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -39,7 +42,12 @@ var ( ) // Controller owns the lifecycle of in-flight reservations. -type Controller struct{ client.Client } +type Controller struct { + client.Client + + // VMClient is a client to call the source of truth for VMs. + VMClient VMClient +} // Reconcile is part of the main kubernetes reconciliation loop which aims to // move the current state of the cluster closer to the desired state. @@ -135,12 +143,83 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return ctrl.Result{RequeueAfter: 10 * time.Second}, nil } - // We don't care where the instance came up. Even if this in-flight - // reservation is for another host, we can prune it. - log.Info("Instance has spawned on a hypervisor, removing in-flight reservation", + // We cannot free this reservation if the instance is currently being + // resized (=migrated to the same hypervisor), i.e. the reservation + // doesn't match the actual size of the vm yet. To check the vm size, + // we need to query the source of truth for vms. + if slices.Contains([]v1alpha1.SchedulingIntent{ + novaapi.RebuildIntent, + novaapi.ResizeIntent, + }, obj.Spec.InFlightReservation.Intent) { + vmSize, err := c.VMClient.GetCurrentVMSize(ctx, obj.Spec.InFlightReservation.VMID) + if err != nil { + log.Error(err, "Failed to get current VM size from vmClient", + "vmID", obj.Spec.InFlightReservation.VMID) + return ctrl.Result{}, err + } + if !reflect.DeepEqual(vmSize, obj.Spec.Resources) { + log.V(1).Info("VM size does not match reservation size yet, requeuing", + "vmID", obj.Spec.InFlightReservation.VMID, + "reservationSize", obj.Spec.Resources, + "currentVMSize", vmSize) + orig := obj.DeepCopy() + meta.SetStatusCondition(&obj.Status.Conditions, metav1.Condition{ + Type: v1alpha1.ReservationConditionReady, + Status: metav1.ConditionUnknown, + Reason: "VMSizeMismatch", + Message: "The current VM size does not match the reservation size yet", + }) + if err := c.Status().Patch(ctx, obj, client.MergeFrom(orig)); err != nil { + log.Error(err, "Failed to update reservation status") + return ctrl.Result{}, err + } + return ctrl.Result{RequeueAfter: 10 * time.Second}, nil + } + } + + // If the instance spawned on the hypervisor expected by the reservation, + // we batch-delete all reservations for that instance, including the + // reconciled one. + if hypervisorName == obj.Spec.TargetHost { + log.V(1).Info("Instance has spawned on the expected hypervisor, deleting reservations", + "vmID", obj.Spec.InFlightReservation.VMID, + "hypervisor", hypervisorName) + reservations := new(v1alpha1.ReservationList) + if err := c.List(ctx, reservations, client.MatchingFields{ + idxReservationByTargetHost: hypervisorName, + }); err != nil { + log.Error(err, "Failed to list reservations for hypervisor", + "hypervisor", hypervisorName) + return ctrl.Result{}, err + } + for _, res := range reservations.Items { + if res.Spec.InFlightReservation == nil { + continue // Not an in-flight reservation, skip. + } + if res.Spec.InFlightReservation.VMID != obj.Spec.InFlightReservation.VMID { + continue // Not the same instance, skip. + } + if err := c.Delete(ctx, &res); err != nil { + log.Error(err, "Failed to delete reservation", + "reservation", res.Name, + "vmID", res.Spec.InFlightReservation.VMID) + return ctrl.Result{}, err + } + log.V(1).Info("Deleted reservation", + "reservation", res.Name, + "vmID", res.Spec.InFlightReservation.VMID) + } + return ctrl.Result{}, nil + } + + // This reservation will be deleted by the hypervisor operator when the + // instance spawns on the expected hypervisor, so we don't need to do + // anything else here. + log.V(1).Info("Reservation stale -- awaiting deletion", "vmID", obj.Spec.InFlightReservation.VMID, - "hypervisor", hypervisorName) - return ctrl.Result{}, c.Delete(ctx, obj) + "expectedHypervisor", obj.Spec.TargetHost, + "actualHypervisor", hypervisorName) + return ctrl.Result{}, nil } // handleReservations generates a new event handler for in flight reservations. diff --git a/internal/scheduling/reservations/inflight/controller_test.go b/internal/scheduling/reservations/inflight/controller_test.go index 9cf6f26d7..44f535efe 100644 --- a/internal/scheduling/reservations/inflight/controller_test.go +++ b/internal/scheduling/reservations/inflight/controller_test.go @@ -5,12 +5,15 @@ package inflight import ( "context" + "errors" "testing" "time" + novaapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/v1alpha1" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" @@ -22,6 +25,17 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" ) +// stubVMClient is a test double for vmClient. If err is non-nil, GetCurrentVMSize +// returns it; otherwise it returns size. +type stubVMClient struct { + size map[hv1.ResourceName]resource.Quantity + err error +} + +func (s *stubVMClient) GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) { + return s.size, s.err +} + // newTestScheme returns a runtime.Scheme with all required types registered. func newTestScheme(t *testing.T) *runtime.Scheme { t.Helper() @@ -172,10 +186,12 @@ func TestReconcile_InstanceNotSpawnedRequeues(t *testing.T) { assertReadyCondition(t, k8sClient, "res-1", metav1.ConditionUnknown, "InstanceNotFound") } -func TestReconcile_InstanceSpawnedDeletesReservation(t *testing.T) { +func TestReconcile_InstanceOnDifferentHostAwaitsDeletion(t *testing.T) { + // Instance landed on a *different* host than the target. The reservation + // is now stale but the hypervisor operator (not this controller) removes + // it, so Reconcile should be a no-op that leaves the reservation intact. scheme := newTestScheme(t) res := newInFlightReservation("res-1", "vm-uuid-1", "host-1") - // Instance landed on a *different* host than the target — controller still deletes. hv1Obj := newHypervisor("host-1") hv2Obj := newHypervisor("host-2", "vm-uuid-1") k8sClient := newTestClient(scheme, res, hv1Obj, hv2Obj) @@ -188,13 +204,13 @@ func TestReconcile_InstanceSpawnedDeletesReservation(t *testing.T) { t.Fatalf("Reconcile returned error: %v", err) } if result.RequeueAfter != 0 { - t.Errorf("expected empty result after deletion, got %+v", result) + t.Errorf("expected empty result, got %+v", result) } + // Reservation still exists — it's waiting for the hypervisor operator. var got v1alpha1.Reservation - err = k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got) - if err == nil { - t.Fatal("expected reservation to be deleted, but Get succeeded") + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("expected reservation to still exist, got error: %v", err) } } @@ -217,6 +233,198 @@ func TestReconcile_InstanceOnTargetHostDeletesReservation(t *testing.T) { } } +func TestReconcile_InstanceOnTargetHostBatchDeletesReservationsForSameVM(t *testing.T) { + // Multiple in-flight reservations pointing at the same target host for the + // same VM (e.g. left over from earlier scheduling attempts). Once the + // instance is confirmed on that host, all of them must be cleaned up in a + // single reconcile. + scheme := newTestScheme(t) + res1 := newInFlightReservation("res-1", "vm-uuid-1", "host-1") + res2 := newInFlightReservation("res-2", "vm-uuid-1", "host-1") + // A reservation for a different VM on the same host must be left alone. + other := newInFlightReservation("res-other", "vm-uuid-2", "host-1") + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res1, res2, other, hv) + c := &Controller{Client: k8sClient} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err == nil { + t.Fatal("expected res-1 to be deleted, but Get succeeded") + } + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-2"}, &got); err == nil { + t.Fatal("expected res-2 to be deleted, but Get succeeded") + } + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-other"}, &got); err != nil { + t.Fatalf("expected res-other to survive, got error: %v", err) + } +} + +// newResizeReservation builds an in-flight reservation with the given intent +// and resource requirements. Used to exercise the resize/rebuild size-check +// branch of Reconcile. +// +//nolint:unparam +func newResizeReservation(name, vmID, targetHost string, intent v1alpha1.SchedulingIntent, resources map[hv1.ResourceName]resource.Quantity) *v1alpha1.Reservation { + return &v1alpha1.Reservation{ + ObjectMeta: metav1.ObjectMeta{Name: name}, + Spec: v1alpha1.ReservationSpec{ + Type: v1alpha1.ReservationTypeInFlight, + SchedulingDomain: v1alpha1.SchedulingDomainNova, + TargetHost: targetHost, + Resources: resources, + InFlightReservation: &v1alpha1.InFlightReservationSpec{ + VMID: vmID, + Intent: intent, + }, + }, + } +} + +func TestReconcile_ResizeSizeMismatchRequeues(t *testing.T) { + // For a resize/rebuild reservation, the instance being present on the + // target host isn't sufficient — the VM must have grown/shrunk to the + // reserved size. Until it has, the controller must requeue and set the + // Ready condition to VMSizeMismatch. + scheme := newTestScheme(t) + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + "memory": resource.MustParse("8Gi"), + } + current := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("2"), + "memory": resource.MustParse("4Gi"), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.ResizeIntent, reserved) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient, VMClient: &stubVMClient{size: current}} + + result, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 10*time.Second { + t.Errorf("RequeueAfter = %v, want 10s", result.RequeueAfter) + } + + // Reservation must still exist and carry the VMSizeMismatch condition. + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("reservation was unexpectedly deleted: %v", err) + } + assertReadyCondition(t, k8sClient, "res-1", metav1.ConditionUnknown, "VMSizeMismatch") +} + +func TestReconcile_ResizeSizeMatchesDeletesReservation(t *testing.T) { + // Once the VM has been resized to the reserved dimensions, the resize + // reservation can be freed like a normal in-flight reservation. + scheme := newTestScheme(t) + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + "memory": resource.MustParse("8Gi"), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.ResizeIntent, reserved) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient, VMClient: &stubVMClient{size: reserved}} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err == nil { + t.Fatal("expected reservation to be deleted, but Get succeeded") + } +} + +func TestReconcile_RebuildSizeMismatchRequeues(t *testing.T) { + // Same branch as resize but exercised via the rebuild intent to make sure + // both intents actually trip the vmClient check. + scheme := newTestScheme(t) + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + } + current := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("2"), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.RebuildIntent, reserved) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient, VMClient: &stubVMClient{size: current}} + + result, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 10*time.Second { + t.Errorf("RequeueAfter = %v, want 10s", result.RequeueAfter) + } + assertReadyCondition(t, k8sClient, "res-1", metav1.ConditionUnknown, "VMSizeMismatch") +} + +func TestReconcile_ResizeVMClientErrorReturnsError(t *testing.T) { + // If the source of truth for VMs can't be reached we must surface the + // error so controller-runtime backs off — silently deleting the + // reservation here would be a resource-accounting bug. + scheme := newTestScheme(t) + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.ResizeIntent, reserved) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + vmErr := errors.New("vm client boom") + c := &Controller{Client: k8sClient, VMClient: &stubVMClient{err: vmErr}} + + _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if !errors.Is(err, vmErr) { + t.Fatalf("Reconcile err = %v, want %v", err, vmErr) + } + + // Reservation must survive the error so it can be retried. + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("reservation was unexpectedly deleted: %v", err) + } +} + +func TestReconcile_NonResizeIntentSkipsVMClient(t *testing.T) { + // For non-resize/rebuild intents the vmClient must not be consulted at + // all — we leave it nil to make an accidental call panic loudly. + scheme := newTestScheme(t) + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.LiveMigrationIntent, + map[hv1.ResourceName]resource.Quantity{"cpu": resource.MustParse("4")}) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient, VMClient: nil} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err == nil { + t.Fatal("expected reservation to be deleted, but Get succeeded") + } +} + func TestIdxReservationByTargetHostFn(t *testing.T) { tests := []struct { name string diff --git a/internal/scheduling/reservations/inflight/vm_client.go b/internal/scheduling/reservations/inflight/vm_client.go new file mode 100644 index 000000000..644575433 --- /dev/null +++ b/internal/scheduling/reservations/inflight/vm_client.go @@ -0,0 +1,96 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package inflight + +import ( + "context" + "log/slog" + "net/http" + + "k8s.io/apimachinery/pkg/api/resource" + "sigs.k8s.io/controller-runtime/pkg/client" + + "github.com/cobaltcore-dev/cortex/pkg/keystone" + "github.com/cobaltcore-dev/cortex/pkg/sso" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "github.com/gophercloud/gophercloud/v2" + "github.com/gophercloud/gophercloud/v2/openstack/compute/v2/servers" + corev1 "k8s.io/api/core/v1" +) + +type VMClient interface { + // GetCurrentVMSize returns the size of the VM with the given ID, + // or an error if the VM cannot be found. + GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) +} + +type novaVMClient struct { + // sc is the service client for the OpenStack Nova API. + sc *gophercloud.ServiceClient +} + +type NovaVMClientConfig 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"` +} + +func InitNovaVMClient(ctx context.Context, client client.Client, config NovaVMClientConfig) (VMClient, error) { + var authenticatedHTTP = http.DefaultClient + if config.SSOSecretRef != nil { + var err error + authenticatedHTTP, err = sso.Connector{Client: client}. + FromSecretRef(ctx, *config.SSOSecretRef) + if err != nil { + return nil, err + } + } + authenticatedKeystone, err := keystone. + Connector{Client: client, HTTPClient: authenticatedHTTP}. + FromSecretRef(ctx, config.KeystoneSecretRef) + if err != nil { + return nil, err + } + // Automatically fetch the nova endpoint from the keystone service catalog. + provider := authenticatedKeystone.Client() + serviceType := "compute" + url, err := authenticatedKeystone.FindEndpoint( + authenticatedKeystone.Availability(), serviceType, + ) + if err != nil { + return nil, err + } + slog.Info("using nova endpoint", "url", url) + sc := &gophercloud.ServiceClient{ + ProviderClient: provider, + Endpoint: url, + Type: serviceType, + // Since microversion 2.53, the hypervisor id and service id is a UUID. + // We need that to find placement resource providers for hypervisors. + Microversion: "2.53", + } + return &novaVMClient{sc: sc}, nil +} + +// GetCurrentVMSize returns the size of the VM with the given ID, or an +// error if the VM cannot be found. +func (c *novaVMClient) GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) { + var server struct { + Flavor struct { + RAM int64 `json:"ram"` + VCPUs int64 `json:"vcpus"` + } `json:"flavor"` + } + err := servers.Get(ctx, c.sc, vmID).ExtractInto(&server) + if err != nil { + return nil, err + } + return map[hv1.ResourceName]resource.Quantity{ + hv1.ResourceCPU: *resource. + NewQuantity(server.Flavor.VCPUs, resource.DecimalSI), + hv1.ResourceMemory: *resource. + NewQuantity(server.Flavor.RAM*1024*1024, resource.BinarySI), + }, nil +} diff --git a/internal/scheduling/reservations/inflight/vm_client_test.go b/internal/scheduling/reservations/inflight/vm_client_test.go new file mode 100644 index 000000000..d8488fe05 --- /dev/null +++ b/internal/scheduling/reservations/inflight/vm_client_test.go @@ -0,0 +1,167 @@ +// Copyright SAP SE +// SPDX-License-Identifier: Apache-2.0 + +package inflight + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + + "github.com/cobaltcore-dev/cortex/pkg/keystone" + testlibKeystone "github.com/cobaltcore-dev/cortex/pkg/keystone/testing" + hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" + "github.com/gophercloud/gophercloud/v2" + "k8s.io/apimachinery/pkg/api/resource" +) + +// setupNovaMockServer starts an httptest server backed by handler and returns +// a mock keystone client whose provider client can drive the gophercloud +// service client at that server's URL. +func setupNovaMockServer(handler http.HandlerFunc) (*httptest.Server, keystone.KeystoneClient) { + server := httptest.NewServer(handler) + return server, &testlibKeystone.MockKeystoneClient{Url: server.URL + "/"} +} + +// newTestNovaVMClient wires a novaVMClient against the given test server and +// keystone mock. +func newTestNovaVMClient(server *httptest.Server, k keystone.KeystoneClient) *novaVMClient { + return &novaVMClient{ + sc: &gophercloud.ServiceClient{ + ProviderClient: k.Client(), + Endpoint: server.URL + "/", + Type: "compute", + Microversion: "2.53", + }, + } +} + +func TestNovaVMClient_GetCurrentVMSize(t *testing.T) { + const vmID = "vm-abc" + handler := func(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + t.Fatalf("expected GET method, got %s", r.Method) + } + if !strings.HasSuffix(r.URL.Path, "/servers/"+vmID) { + t.Fatalf("unexpected path: %s", r.URL.Path) + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, err := w.Write([]byte(`{"server": {"id": "vm-abc", "flavor": {"ram": 2048, "vcpus": 4}}}`)) + if err != nil { + t.Fatalf("failed to write response: %v", err) + } + } + server, k := setupNovaMockServer(handler) + defer server.Close() + c := newTestNovaVMClient(server, k) + + size, err := c.GetCurrentVMSize(t.Context(), vmID) + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + // CPU is DecimalSI-encoded quantity of vcpus. + cpu, ok := size[hv1.ResourceCPU] + if !ok { + t.Fatalf("expected CPU resource in size map: %+v", size) + } + if got := cpu.Value(); got != 4 { + t.Errorf("expected 4 vCPUs, got %d", got) + } + // Memory is reported in bytes: ram (MiB) * 1024 * 1024. + mem, ok := size[hv1.ResourceMemory] + if !ok { + t.Fatalf("expected Memory resource in size map: %+v", size) + } + wantMem := int64(2048) * 1024 * 1024 + if got := mem.Value(); got != wantMem { + t.Errorf("expected %d bytes memory, got %d", wantMem, got) + } + if mem.Format != resource.BinarySI { + t.Errorf("expected BinarySI format for memory, got %v", mem.Format) + } + if cpu.Format != resource.DecimalSI { + t.Errorf("expected DecimalSI format for cpu, got %v", cpu.Format) + } +} + +func TestNovaVMClient_GetCurrentVMSize_ZeroValues(t *testing.T) { + // Server that returns a flavor with zero ram/vcpus — we still expect + // a valid, non-nil size map (values are zero) and no error. + handler := func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, err := w.Write([]byte(`{"server": {"id": "vm-zero", "flavor": {"ram": 0, "vcpus": 0}}}`)) + if err != nil { + t.Fatalf("failed to write response: %v", err) + } + } + server, k := setupNovaMockServer(handler) + defer server.Close() + c := newTestNovaVMClient(server, k) + + size, err := c.GetCurrentVMSize(t.Context(), "vm-zero") + if err != nil { + t.Fatalf("expected no error, got %v", err) + } + cpu := size[hv1.ResourceCPU] + if got := cpu.Value(); got != 0 { + t.Errorf("expected 0 vCPUs, got %d", got) + } + mem := size[hv1.ResourceMemory] + if got := mem.Value(); got != 0 { + t.Errorf("expected 0 bytes memory, got %d", got) + } +} + +func TestNovaVMClient_GetCurrentVMSize_NotFound(t *testing.T) { + handler := func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusNotFound) + _, err := w.Write([]byte(`{"itemNotFound": {"message": "Instance not found", "code": 404}}`)) + if err != nil { + t.Fatalf("failed to write response: %v", err) + } + } + server, k := setupNovaMockServer(handler) + defer server.Close() + c := newTestNovaVMClient(server, k) + + _, err := c.GetCurrentVMSize(t.Context(), "missing-vm") + if err == nil { + t.Fatal("expected error for 404, got nil") + } +} + +func TestNovaVMClient_GetCurrentVMSize_ServerError(t *testing.T) { + handler := func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusInternalServerError) + } + server, k := setupNovaMockServer(handler) + defer server.Close() + c := newTestNovaVMClient(server, k) + + _, err := c.GetCurrentVMSize(t.Context(), "any-vm") + if err == nil { + t.Fatal("expected error for 500, got nil") + } +} + +func TestNovaVMClient_GetCurrentVMSize_MalformedJSON(t *testing.T) { + handler := func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusOK) + _, err := w.Write([]byte(`not a json`)) + if err != nil { + t.Fatalf("failed to write response: %v", err) + } + } + server, k := setupNovaMockServer(handler) + defer server.Close() + c := newTestNovaVMClient(server, k) + + _, err := c.GetCurrentVMSize(t.Context(), "any-vm") + if err == nil { + t.Fatal("expected error for malformed JSON, got nil") + } +} From 12bdbafa4bb9924bfe920c4f8fbbfc2f436ee022 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Tue, 7 Jul 2026 16:36:09 +0200 Subject: [PATCH 7/9] Polishing & fixing --- cmd/manager/main.go | 12 +--- .../reservations/inflight/controller.go | 10 +++- .../reservations/inflight/controller_test.go | 6 +- .../reservations/inflight/vm_client.go | 56 +++++++++++++------ 4 files changed, 57 insertions(+), 27 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 71ecc3e3d..d609c25b8 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -480,15 +480,9 @@ func main() { setupLog.Info("enabling controller", "controller", "inflight-reservation-controller") config := conf.GetConfigOrDie[inflight.NovaVMClientConfig]() - novaVMClient, err := inflight.InitNovaVMClient(ctx, multiclusterClient, config) - if err != nil { - setupLog.Error(err, "unable to initialize nova VM client") - os.Exit(1) - } - if err := (&inflight.Controller{ - Client: multiclusterClient, - VMClient: novaVMClient, - }).SetupWithManager(ctx, mgr); err != nil { + vmClient := inflight.NewNovaVMClient(config) + controller := &inflight.Controller{Client: multiclusterClient, VMClient: vmClient} + if err := controller.SetupWithManager(ctx, mgr); err != nil { setupLog.Error(err, "unable to create controller", "controller", "inflight-reservation-controller") os.Exit(1) diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index a19c30eca..2838da2e4 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -23,6 +23,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/event" "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/manager" "sigs.k8s.io/controller-runtime/pkg/predicate" "sigs.k8s.io/controller-runtime/pkg/reconcile" ) @@ -355,11 +356,18 @@ func (c *Controller) predicateHypervisors() predicate.Predicate { // Reservation CRD across all clusters and trigger reconciliations accordingly. func (c *Controller) SetupWithManager(ctx context.Context, mgr ctrl.Manager) (err error) { // Check that the provided client is a multicluster client, since we need - // that to watch for hypervisors across clusters. + // that to watch for hypervisors across clusters. Do this before adding + // any runnables so a misconfigured setup fails fast. mcl, ok := c.Client.(*multicluster.Client) if !ok { return errors.New("provided client must be a multicluster client") } + // Add the vm client as runnable to the manager. + if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { + return c.VMClient.StartWithKubernetesSecrets(ctx, c.Client) + })); err != nil { + return err + } bldr := multicluster.BuildController(mcl, mgr) // The reservation crd & hypervisor crd may be distributed across multiple // remote clusters. diff --git a/internal/scheduling/reservations/inflight/controller_test.go b/internal/scheduling/reservations/inflight/controller_test.go index 44f535efe..b4b0cfca9 100644 --- a/internal/scheduling/reservations/inflight/controller_test.go +++ b/internal/scheduling/reservations/inflight/controller_test.go @@ -32,6 +32,10 @@ type stubVMClient struct { err error } +func (s *stubVMClient) StartWithKubernetesSecrets(ctx context.Context, client client.Client) error { + return nil +} + func (s *stubVMClient) GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) { return s.size, s.err } @@ -641,7 +645,7 @@ func TestHandleHypervisors_NoMatchingReservations(t *testing.T) { func TestSetupWithManager_RejectsNonMulticlusterClient(t *testing.T) { scheme := newTestScheme(t) - c := &Controller{Client: newTestClient(scheme)} + c := &Controller{Client: newTestClient(scheme), VMClient: &stubVMClient{}} err := c.SetupWithManager(context.Background(), nil) if err == nil { t.Fatal("expected error for non-multicluster client, got nil") diff --git a/internal/scheduling/reservations/inflight/vm_client.go b/internal/scheduling/reservations/inflight/vm_client.go index 644575433..bcdcdebe7 100644 --- a/internal/scheduling/reservations/inflight/vm_client.go +++ b/internal/scheduling/reservations/inflight/vm_client.go @@ -5,27 +5,30 @@ package inflight import ( "context" - "log/slog" + "errors" "net/http" - "k8s.io/apimachinery/pkg/api/resource" - "sigs.k8s.io/controller-runtime/pkg/client" - "github.com/cobaltcore-dev/cortex/pkg/keystone" "github.com/cobaltcore-dev/cortex/pkg/sso" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "github.com/gophercloud/gophercloud/v2" "github.com/gophercloud/gophercloud/v2/openstack/compute/v2/servers" corev1 "k8s.io/api/core/v1" + "k8s.io/apimachinery/pkg/api/resource" + ctrl "sigs.k8s.io/controller-runtime" + "sigs.k8s.io/controller-runtime/pkg/client" ) type VMClient interface { + StartWithKubernetesSecrets(ctx context.Context, client client.Client) error // GetCurrentVMSize returns the size of the VM with the given ID, // or an error if the VM cannot be found. GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) } type novaVMClient struct { + // config is the configuration for the novaVMClient, including keystone and SSO credentials. + config NovaVMClientConfig // sc is the service client for the OpenStack Nova API. sc *gophercloud.ServiceClient } @@ -37,22 +40,34 @@ type NovaVMClientConfig struct { SSOSecretRef *corev1.SecretReference `json:"ssoSecretRef"` } -func InitNovaVMClient(ctx context.Context, client client.Client, config NovaVMClientConfig) (VMClient, error) { +func NewNovaVMClient(config NovaVMClientConfig) VMClient { + return &novaVMClient{ + config: config, + } +} + +func (c *novaVMClient) StartWithKubernetesSecrets(ctx context.Context, client client.Client) error { + log := ctrl.LoggerFrom(ctx) + log.Info("starting novaVMClient with Kubernetes secrets") var authenticatedHTTP = http.DefaultClient - if config.SSOSecretRef != nil { + if c.config.SSOSecretRef != nil { var err error authenticatedHTTP, err = sso.Connector{Client: client}. - FromSecretRef(ctx, *config.SSOSecretRef) + FromSecretRef(ctx, *c.config.SSOSecretRef) if err != nil { - return nil, err + log.Error(err, "failed to create SSO authenticated HTTP client") + return err } + log.Info("successfully created SSO authenticated HTTP client") } authenticatedKeystone, err := keystone. Connector{Client: client, HTTPClient: authenticatedHTTP}. - FromSecretRef(ctx, config.KeystoneSecretRef) + FromSecretRef(ctx, c.config.KeystoneSecretRef) if err != nil { - return nil, err + log.Error(err, "failed to create authenticated keystone client") + return err } + log.Info("successfully created authenticated keystone client") // Automatically fetch the nova endpoint from the keystone service catalog. provider := authenticatedKeystone.Client() serviceType := "compute" @@ -60,10 +75,11 @@ func InitNovaVMClient(ctx context.Context, client client.Client, config NovaVMCl authenticatedKeystone.Availability(), serviceType, ) if err != nil { - return nil, err + log.Error(err, "failed to find nova endpoint in keystone service catalog") + return err } - slog.Info("using nova endpoint", "url", url) - sc := &gophercloud.ServiceClient{ + log.Info("successfully found nova endpoint in keystone service catalog", "url", url) + c.sc = &gophercloud.ServiceClient{ ProviderClient: provider, Endpoint: url, Type: serviceType, @@ -71,12 +87,17 @@ func InitNovaVMClient(ctx context.Context, client client.Client, config NovaVMCl // We need that to find placement resource providers for hypervisors. Microversion: "2.53", } - return &novaVMClient{sc: sc}, nil + return nil } // GetCurrentVMSize returns the size of the VM with the given ID, or an // error if the VM cannot be found. func (c *novaVMClient) GetCurrentVMSize(ctx context.Context, vmID string) (map[hv1.ResourceName]resource.Quantity, error) { + log := ctrl.LoggerFrom(ctx) + if c.sc == nil { + log.Error(nil, "nova service client not initialized yet") + return nil, errors.New("nova service client not initialized yet") + } var server struct { Flavor struct { RAM int64 `json:"ram"` @@ -85,12 +106,15 @@ func (c *novaVMClient) GetCurrentVMSize(ctx context.Context, vmID string) (map[h } err := servers.Get(ctx, c.sc, vmID).ExtractInto(&server) if err != nil { + log.Error(err, "failed to get server details from nova", "vmID", vmID) return nil, err } - return map[hv1.ResourceName]resource.Quantity{ + size := map[hv1.ResourceName]resource.Quantity{ hv1.ResourceCPU: *resource. NewQuantity(server.Flavor.VCPUs, resource.DecimalSI), hv1.ResourceMemory: *resource. NewQuantity(server.Flavor.RAM*1024*1024, resource.BinarySI), - }, nil + } + log.Info("successfully retrieved VM size from nova", "vmID", vmID, "size", size) + return size, nil } From 71be3ac933c3bf5d2eb368dffd5978e27810362c Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Tue, 7 Jul 2026 16:39:37 +0200 Subject: [PATCH 8/9] Address CodeRabbit review: semantic Quantity compare & host-guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - controller.go: replace reflect.DeepEqual on the reservation resource map with a per-key resource.Quantity Cmp check. reflect.DeepEqual on Quantity can return false for numerically-equal values because Quantity's unexported cached string/format state differs between values from API decoding and values freshly constructed by the vm client — that would leave resize/rebuild reservations stuck in VMSizeMismatch forever. - controller.go: gate the resize/rebuild size-check on the instance landing on the reservation's TargetHost. For a stale reservation (instance found on a different host) the previous code fired an unnecessary VMClient call and marked the reservation VMSizeMismatch instead of falling through to the awaiting-deletion no-op. - controller_test.go: add regression tests covering both fixes. --- .../reservations/inflight/controller.go | 27 ++++++- .../reservations/inflight/controller_test.go | 74 +++++++++++++++++++ 2 files changed, 98 insertions(+), 3 deletions(-) diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index 2838da2e4..9a55e6bfa 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -147,8 +147,13 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu // We cannot free this reservation if the instance is currently being // resized (=migrated to the same hypervisor), i.e. the reservation // doesn't match the actual size of the vm yet. To check the vm size, - // we need to query the source of truth for vms. - if slices.Contains([]v1alpha1.SchedulingIntent{ + // we need to query the source of truth for vms. Only relevant when the + // instance actually landed on the reservation's target host — for a + // stale reservation (instance on a different host) this branch would + // otherwise fire an unnecessary vmClient call and mark the reservation + // VMSizeMismatch when it should just fall through to the + // awaiting-deletion no-op below. + if hypervisorName == obj.Spec.TargetHost && slices.Contains([]v1alpha1.SchedulingIntent{ novaapi.RebuildIntent, novaapi.ResizeIntent, }, obj.Spec.InFlightReservation.Intent) { @@ -158,7 +163,23 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu "vmID", obj.Spec.InFlightReservation.VMID) return ctrl.Result{}, err } - if !reflect.DeepEqual(vmSize, obj.Spec.Resources) { + // Compare the resource maps semantically: reflect.DeepEqual on + // resource.Quantity can return false for numerically equal values + // because Quantity's unexported cached string/format state may + // differ between values coming from API decoding (reservation + // spec) and values freshly constructed by the vm client. Using + // Quantity.Cmp avoids getting stuck in VMSizeMismatch forever. + sizeMatches := len(vmSize) == len(obj.Spec.Resources) + if sizeMatches { + for k, want := range obj.Spec.Resources { + got, ok := vmSize[k] + if !ok || got.Cmp(want) != 0 { + sizeMatches = false + break + } + } + } + if !sizeMatches { log.V(1).Info("VM size does not match reservation size yet, requeuing", "vmID", obj.Spec.InFlightReservation.VMID, "reservationSize", obj.Spec.Resources, diff --git a/internal/scheduling/reservations/inflight/controller_test.go b/internal/scheduling/reservations/inflight/controller_test.go index b4b0cfca9..6f61d3a76 100644 --- a/internal/scheduling/reservations/inflight/controller_test.go +++ b/internal/scheduling/reservations/inflight/controller_test.go @@ -352,6 +352,80 @@ func TestReconcile_ResizeSizeMatchesDeletesReservation(t *testing.T) { } } +// TestReconcile_ResizeSizeMatchesSemanticallyDeletesReservation guards the +// semantic quantity comparison in Reconcile: the reservation spec and the vm +// client return numerically-identical quantities that were constructed +// differently (MustParse vs NewQuantity), so reflect.DeepEqual returns false +// even though the resources match. If the controller ever regresses to +// reflect.DeepEqual, this test starts failing. +func TestReconcile_ResizeSizeMatchesSemanticallyDeletesReservation(t *testing.T) { + scheme := newTestScheme(t) + // Reservation spec side: built like the API server would (MustParse). + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + "memory": resource.MustParse("8Gi"), + } + // VM client side: built like the nova vm client does (NewQuantity). + current := map[hv1.ResourceName]resource.Quantity{ + "cpu": *resource.NewQuantity(4, resource.DecimalSI), + "memory": *resource.NewQuantity(8*1024*1024*1024, resource.BinarySI), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.ResizeIntent, reserved) + hv := newHypervisor("host-1", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv) + c := &Controller{Client: k8sClient, VMClient: &stubVMClient{size: current}} + + if _, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }); err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err == nil { + t.Fatal("expected reservation to be deleted (semantic size match), but Get succeeded") + } +} + +// TestReconcile_ResizeStaleReservationSkipsVMClient covers the guard that +// stops the size-check branch from running when the instance landed on a host +// other than the reservation's target. A stale resize reservation should fall +// through to the awaiting-deletion no-op — not fire a vmClient call and get +// marked VMSizeMismatch. VMClient is left nil so an accidental call panics. +func TestReconcile_ResizeStaleReservationSkipsVMClient(t *testing.T) { + scheme := newTestScheme(t) + reserved := map[hv1.ResourceName]resource.Quantity{ + "cpu": resource.MustParse("4"), + } + res := newResizeReservation("res-1", "vm-uuid-1", "host-1", novaapi.ResizeIntent, reserved) + // Instance ended up on host-2, not the target host-1. + hv1Obj := newHypervisor("host-1") + hv2Obj := newHypervisor("host-2", "vm-uuid-1") + k8sClient := newTestClient(scheme, res, hv1Obj, hv2Obj) + c := &Controller{Client: k8sClient, VMClient: nil} + + result, err := c.Reconcile(context.Background(), ctrl.Request{ + NamespacedName: types.NamespacedName{Name: "res-1"}, + }) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 0 { + t.Errorf("expected empty result for stale reservation, got %+v", result) + } + + // Reservation must still exist and must NOT have been marked with + // VMSizeMismatch — it's simply awaiting the hypervisor operator to + // prune it. + var got v1alpha1.Reservation + if err := k8sClient.Get(context.Background(), types.NamespacedName{Name: "res-1"}, &got); err != nil { + t.Fatalf("expected reservation to still exist, got error: %v", err) + } + if cond := meta.FindStatusCondition(got.Status.Conditions, v1alpha1.ReservationConditionReady); cond != nil && cond.Reason == "VMSizeMismatch" { + t.Errorf("stale reservation was incorrectly marked VMSizeMismatch") + } +} + func TestReconcile_RebuildSizeMismatchRequeues(t *testing.T) { // Same branch as resize but exercised via the rebuild intent to make sure // both intents actually trip the vmClient check. From 27718bba1189bcbc77948eb6758faa25b0335133 Mon Sep 17 00:00:00 2001 From: Philipp Matthes Date: Thu, 9 Jul 2026 07:44:21 +0200 Subject: [PATCH 9/9] Small code comment refinement --- internal/scheduling/reservations/inflight/controller.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/scheduling/reservations/inflight/controller.go b/internal/scheduling/reservations/inflight/controller.go index 9a55e6bfa..82c9ac8ea 100644 --- a/internal/scheduling/reservations/inflight/controller.go +++ b/internal/scheduling/reservations/inflight/controller.go @@ -234,7 +234,7 @@ func (c *Controller) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Resu return ctrl.Result{}, nil } - // This reservation will be deleted by the hypervisor operator when the + // This reservation will be deleted by the controller when the // instance spawns on the expected hypervisor, so we don't need to do // anything else here. log.V(1).Info("Reservation stale -- awaiting deletion",