From fddef47bd8d846012ef626745f70ba2506df56f1 Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Wed, 8 Jul 2026 13:58:30 +0200 Subject: [PATCH 1/8] refactor --- cmd/manager/main.go | 9 +- .../reservations/capacity/config.go | 23 +++- .../reservations/capacity/controller.go | 118 +++++++++++++----- 3 files changed, 110 insertions(+), 40 deletions(-) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index af86e3d84..2ffeb0f55 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -771,16 +771,15 @@ func main() { setupLog.Error(err, "failed to register capacity monitor metrics, continuing without metrics") } - capacityController := capacity.NewController(multiclusterClient, capacityConfig, commitmentsVMSource) - if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error { - return capacityController.Start(ctx) - })); err != nil { - setupLog.Error(err, "unable to add capacity controller to manager") + if err := capacity.NewController(multiclusterClient, capacityConfig, commitmentsVMSource). + SetupWithManager(mgr, multiclusterClient); err != nil { + setupLog.Error(err, "unable to create controller", "controller", "capacity") os.Exit(1) } setupLog.Info("capacity-controller registered", "schedulerURL", capacityConfig.SchedulerURL, "reconcileInterval", capacityConfig.ReconcileInterval, + "minReconcileInterval", capacityConfig.MinReconcileInterval, "totalPipeline", capacityConfig.TotalPipeline, "placeablePipeline", capacityConfig.PlaceablePipeline) } diff --git a/internal/scheduling/reservations/capacity/config.go b/internal/scheduling/reservations/capacity/config.go index 264a0b59d..cd33875b9 100644 --- a/internal/scheduling/reservations/capacity/config.go +++ b/internal/scheduling/reservations/capacity/config.go @@ -9,11 +9,18 @@ import ( metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" ) -// Config holds configuration for the capacity controller. +// Config holds configuration for the capacity reconciler. type Config struct { - // ReconcileInterval is how often the controller probes the scheduler and updates CRDs. + // ReconcileInterval is the periodic floor: how often the reconciler re-runs even without a + // watch event. Acts as a fallback for changes not covered by watches (e.g. blocked memory drift). ReconcileInterval metav1.Duration `json:"capacityReconcileInterval"` + // MinReconcileInterval is the minimum time between two consecutive reconcile runs. + // If Reconcile() is called sooner than this since the last successful run, it returns early + // with RequeueAfter set to the remaining duration. Prevents back-to-back reconciles on rapid + // watch events (e.g. a batch of CommittedResource updates). + MinReconcileInterval metav1.Duration `json:"capacityMinReconcileInterval"` + // TotalPipeline is the scheduler pipeline used for the empty-state probe. // This pipeline should ignore current VM allocations (e.g. kvm-report-capacity). TotalPipeline string `json:"capacityTotalPipeline"` @@ -32,6 +39,9 @@ func (c *Config) ApplyDefaults() { if c.ReconcileInterval.Duration == 0 { c.ReconcileInterval = defaults.ReconcileInterval } + if c.MinReconcileInterval.Duration == 0 { + c.MinReconcileInterval = defaults.MinReconcileInterval + } if c.TotalPipeline == "" { c.TotalPipeline = defaults.TotalPipeline } @@ -45,9 +55,10 @@ func (c *Config) ApplyDefaults() { func DefaultConfig() Config { return Config{ - ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, - TotalPipeline: "kvm-report-capacity", - PlaceablePipeline: "kvm-general-purpose-load-balancing-no-history", - SchedulerURL: "http://localhost:8080/scheduler/nova/external", + ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, + MinReconcileInterval: metav1.Duration{Duration: 30 * time.Second}, + TotalPipeline: "kvm-report-capacity", + PlaceablePipeline: "kvm-general-purpose-load-balancing-no-history", + SchedulerURL: "http://localhost:8080/scheduler/nova/external", } } diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index 1e69b22cb..8bd4ee01d 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -9,6 +9,7 @@ import ( "hash/fnv" "sort" "strings" + "sync" "time" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" @@ -19,28 +20,40 @@ import ( "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" + "sigs.k8s.io/controller-runtime/pkg/controller" + "sigs.k8s.io/controller-runtime/pkg/handler" + "sigs.k8s.io/controller-runtime/pkg/reconcile" schedulerapi "github.com/cobaltcore-dev/cortex/api/external/nova" "github.com/cobaltcore-dev/cortex/api/scheduling" "github.com/cobaltcore-dev/cortex/api/v1alpha1" "github.com/cobaltcore-dev/cortex/internal/knowledge/extractor/plugins/compute" "github.com/cobaltcore-dev/cortex/internal/scheduling/reservations" + "github.com/cobaltcore-dev/cortex/pkg/multicluster" ) var log = ctrl.Log.WithName("capacity-controller").WithValues("module", "capacity") -// Controller reconciles FlavorGroupCapacity CRDs on a fixed interval. -// For each AZ it probes all flavor groups, runs the round-robin capacity split, then writes -// one FlavorGroupCapacity CRD per (flavor group × AZ) pair. -type Controller struct { +// coalescedKey is the single reconcile key used for all watch events. +// All CRD changes (Knowledge, Hypervisor, CommittedResource, Pipeline) are coalesced +// into this one key so rapid changes never cause rapid-fire scheduler probes. +const coalescedKey = "capacity" + +// Reconciler reconciles FlavorGroupCapacity CRDs, driven by both watch events and a +// periodic floor timer. All watch events are coalesced into a single reconcile key so +// rapid changes produce at most one queued reconcile. +type Reconciler struct { client client.Client vmSource reservations.VMSource schedulerClient *reservations.SchedulerClient config Config + + mu sync.Mutex + lastReconcileAt time.Time } -func NewController(c client.Client, config Config, vmSource reservations.VMSource) *Controller { - return &Controller{ +func NewController(c client.Client, config Config, vmSource reservations.VMSource) *Reconciler { + return &Reconciler{ client: c, vmSource: vmSource, schedulerClient: reservations.NewSchedulerClient(config.SchedulerURL), @@ -48,23 +61,70 @@ func NewController(c client.Client, config Config, vmSource reservations.VMSourc } } -// Start runs the periodic reconcile loop. Implements manager.Runnable. -func (c *Controller) Start(ctx context.Context) error { - timer := time.NewTimer(0) // fire immediately on start - defer timer.Stop() - - for { - select { - case <-ctx.Done(): - return nil - case <-timer.C: - cycleCtx := WithNewGlobalRequestID(ctx) - if err := c.reconcileAll(cycleCtx); err != nil { - LoggerFromContext(cycleCtx).Error(err, "reconcile cycle failed") - } - timer.Reset(c.config.ReconcileInterval.Duration) - } +// Reconcile implements reconcile.Reconciler. It is called by controller-runtime whenever a +// watched CRD changes, and also on the periodic RequeueAfter floor set by ReconcileInterval. +// If called sooner than MinReconcileInterval since the last successful run, it returns early. +func (c *Reconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) { + c.mu.Lock() + elapsed := time.Since(c.lastReconcileAt) + minInterval := c.config.MinReconcileInterval.Duration + c.mu.Unlock() + + if c.lastReconcileAt != (time.Time{}) && elapsed < minInterval { + remaining := minInterval - elapsed + LoggerFromContext(ctx).V(1).Info("skipping reconcile: min interval not elapsed", + "elapsed", elapsed.Round(time.Second), + "remaining", remaining.Round(time.Second)) + return ctrl.Result{RequeueAfter: remaining}, nil + } + + cycleCtx := WithNewGlobalRequestID(ctx) + if err := c.reconcileAll(cycleCtx); err != nil { + LoggerFromContext(cycleCtx).Error(err, "reconcile cycle failed") + return ctrl.Result{}, err + } + + c.mu.Lock() + c.lastReconcileAt = time.Now() + c.mu.Unlock() + + return ctrl.Result{RequeueAfter: c.config.ReconcileInterval.Duration}, nil +} + +// SetupWithManager registers the reconciler with the controller manager and sets up watches +// on all CRDs that affect capacity output. All events are coalesced to a single key. +func (c *Reconciler) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client) error { + log.Info("starting capacity reconciler", + "reconcileInterval", c.config.ReconcileInterval.Duration, + "minReconcileInterval", c.config.MinReconcileInterval.Duration) + + coalesce := func(_ context.Context, _ client.Object) []reconcile.Request { + return []reconcile.Request{{NamespacedName: types.NamespacedName{Name: coalescedKey}}} + } + + bldr := multicluster.BuildController(mcl, mgr) + var err error + + bldr, err = bldr.WatchesMulticluster(&v1alpha1.Knowledge{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + if err != nil { + return fmt.Errorf("failed to watch Knowledge: %w", err) + } + bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + if err != nil { + return fmt.Errorf("failed to watch Hypervisor: %w", err) + } + bldr, err = bldr.WatchesMulticluster(&v1alpha1.CommittedResource{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + if err != nil { + return fmt.Errorf("failed to watch CommittedResource: %w", err) } + bldr, err = bldr.WatchesMulticluster(&v1alpha1.Pipeline{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + if err != nil { + return fmt.Errorf("failed to watch Pipeline: %w", err) + } + + return bldr.Named("capacity"). + WithOptions(controller.Options{MaxConcurrentReconciles: 1}). + Complete(c) } type vmUsageKey struct{ group, az string } @@ -79,7 +139,7 @@ type vmUsage struct { } // reconcileAll iterates all AZs, runs the round-robin split per AZ, then writes CRDs. -func (c *Controller) reconcileAll(ctx context.Context) error { +func (c *Reconciler) reconcileAll(ctx context.Context) error { logger := LoggerFromContext(ctx) startTime := time.Now() @@ -131,7 +191,7 @@ func (c *Controller) reconcileAll(ctx context.Context) error { // computeVMUsage fetches running VMs and aggregates usage per (flavorGroup, az). // On error returns an empty map with fresh=false — callers must not overwrite running fields. -func (c *Controller) computeVMUsage( +func (c *Reconciler) computeVMUsage( ctx context.Context, flavorGroups map[string]compute.FlavorGroupFeature, hvs []hv1.Hypervisor, @@ -233,7 +293,7 @@ func hvRemainingResources(hv hv1.Hypervisor, blockedMemBytes int64) map[string]i // reconcileAZ runs the round-robin capacity split for all flavor groups in one AZ, // then writes one FlavorGroupCapacity CRD per group that had all probes succeed. // Groups with failed probes are skipped — their CRDs retain the last good state. -func (c *Controller) reconcileAZ( +func (c *Reconciler) reconcileAZ( ctx context.Context, az string, flavorGroups map[string]compute.FlavorGroupFeature, @@ -414,7 +474,7 @@ func (c *Controller) reconcileAZ( } // writeCRD upserts one FlavorGroupCapacity CRD with fresh computed values. -func (c *Controller) writeCRD( +func (c *Reconciler) writeCRD( ctx context.Context, groupName string, groupData compute.FlavorGroupFeature, @@ -507,7 +567,7 @@ func (c *Controller) writeCRD( // probeScheduler calls the scheduler and returns slot count, host count, and candidate host names. // ignoreAllocations=true (total probe) uses raw effective capacity; false (placeable probe) subtracts allocations. -func (c *Controller) probeScheduler( +func (c *Reconciler) probeScheduler( ctx context.Context, flavor compute.FlavorInGroup, az, pipeline string, @@ -586,7 +646,7 @@ func (c *Controller) probeScheduler( // blockedMemoryByHost returns total reservation-blocked bytes per host. // Both TargetHost and Status.Host are blocked; migration blocks both simultaneously. -func (c *Controller) blockedMemoryByHost(ctx context.Context) (map[string]int64, error) { +func (c *Reconciler) blockedMemoryByHost(ctx context.Context) (map[string]int64, error) { var list v1alpha1.ReservationList if err := c.client.List(ctx, &list); err != nil { return nil, fmt.Errorf("failed to list reservations: %w", err) @@ -622,7 +682,7 @@ func (c *Controller) blockedMemoryByHost(ctx context.Context) (map[string]int64, // sumCommittedCapacity sums active CommittedResource amounts (memory type, guaranteed/confirmed) // for the given (flavorGroup, az) pair. Returns the total in smallest-flavor slots. -func (c *Controller) sumCommittedCapacity(ctx context.Context, groupName, az string, smallestFlavorBytes int64) (int64, error) { +func (c *Reconciler) sumCommittedCapacity(ctx context.Context, groupName, az string, smallestFlavorBytes int64) (int64, error) { var list v1alpha1.CommittedResourceList if err := c.client.List(ctx, &list); err != nil { return 0, fmt.Errorf("failed to list CommittedResources: %w", err) From 72c3ca834e41d70eb5cd19e5c0cabc61c4c9936b Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Wed, 8 Jul 2026 13:58:33 +0200 Subject: [PATCH 2/8] test --- .../reservations/capacity/controller_test.go | 119 +++++++++++++++++- 1 file changed, 117 insertions(+), 2 deletions(-) diff --git a/internal/scheduling/reservations/capacity/controller_test.go b/internal/scheduling/reservations/capacity/controller_test.go index c33d98be4..31d42d749 100644 --- a/internal/scheduling/reservations/capacity/controller_test.go +++ b/internal/scheduling/reservations/capacity/controller_test.go @@ -11,12 +11,14 @@ import ( "regexp" "sort" "testing" + "time" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" "k8s.io/apimachinery/pkg/api/resource" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" "k8s.io/apimachinery/pkg/runtime" "k8s.io/apimachinery/pkg/types" + ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/client/fake" @@ -111,8 +113,8 @@ func newMockSchedulerServer(t *testing.T, hosts []string) *httptest.Server { })) } -// newController is a test helper that creates a Controller with a nil VMSource. -func newController(t *testing.T, c client.Client, cfg Config) *Controller { +// newController is a test helper that creates a Reconciler with a nil VMSource. +func newController(t *testing.T, c client.Client, cfg Config) *Reconciler { t.Helper() return NewController(c, cfg, nil) } @@ -848,3 +850,116 @@ func TestProbeScheduler_SubtractsReservationBlocksWhenNotIgnored(t *testing.T) { t.Errorf("placeable capacity = %d, want 1 (3 slots − 1 alloc − 1 reservation)", placeableCap) } } + +// TestReconcile_ReactsToKnowledgeChange verifies that Reconcile() runs reconcileAll() and +// writes FlavorGroupCapacity CRDs when triggered by a watch event (simulated here by calling +// Reconcile directly with the coalesced key). +func TestReconcile_ReactsToKnowledgeChange(t *testing.T) { + const ( + groupName = "hana-v2" + az = "qa-de-1a" + memMB = 4096 + memBytes = int64(memMB) * 1024 * 1024 + ) + + scheme := newTestScheme(t) + hv := newHypervisor("host-1", az, memBytes) + knowledge := newFlavorGroupKnowledge(t, groupName, memMB) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(knowledge, hv). + WithStatusSubresource(&v1alpha1.FlavorGroupCapacity{}, &v1alpha1.Knowledge{}). + Build() + + schedulerServer := newMockSchedulerServer(t, []string{"host-1"}) + defer schedulerServer.Close() + + r := NewController(fakeClient, Config{ + SchedulerURL: schedulerServer.URL, + TotalPipeline: "kvm-report-capacity", + PlaceablePipeline: "kvm-general-purpose", + ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, + MinReconcileInterval: metav1.Duration{Duration: 30 * time.Second}, + }, nil) + + req := ctrl.Request{NamespacedName: types.NamespacedName{Name: coalescedKey}} + result, err := r.Reconcile(context.Background(), req) + if err != nil { + t.Fatalf("Reconcile returned error: %v", err) + } + if result.RequeueAfter != 5*time.Minute { + t.Errorf("RequeueAfter = %v, want 5m (periodic floor)", result.RequeueAfter) + } + + // reconcileAll should have written one CRD for the single (group × AZ) pair. + var list v1alpha1.FlavorGroupCapacityList + if err := fakeClient.List(context.Background(), &list); err != nil { + t.Fatalf("failed to list CRDs: %v", err) + } + if len(list.Items) != 1 { + t.Errorf("expected 1 FlavorGroupCapacity CRD after reactive reconcile, got %d", len(list.Items)) + } +} + +// TestReconcile_MinIntervalEarlyReturn verifies that a second Reconcile() call within +// MinReconcileInterval returns early (no reconcileAll) with RequeueAfter set to the +// remaining cooldown duration. +func TestReconcile_MinIntervalEarlyReturn(t *testing.T) { + const ( + groupName = "hana-v2" + az = "qa-de-1a" + memMB = 4096 + memBytes = int64(memMB) * 1024 * 1024 + ) + + scheme := newTestScheme(t) + hv := newHypervisor("host-1", az, memBytes) + knowledge := newFlavorGroupKnowledge(t, groupName, memMB) + + fakeClient := fake.NewClientBuilder(). + WithScheme(scheme). + WithObjects(knowledge, hv). + WithStatusSubresource(&v1alpha1.FlavorGroupCapacity{}, &v1alpha1.Knowledge{}). + Build() + + schedulerServer := newMockSchedulerServer(t, []string{"host-1"}) + defer schedulerServer.Close() + + minInterval := 30 * time.Second + r := NewController(fakeClient, Config{ + SchedulerURL: schedulerServer.URL, + TotalPipeline: "kvm-report-capacity", + PlaceablePipeline: "kvm-general-purpose", + ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, + MinReconcileInterval: metav1.Duration{Duration: minInterval}, + }, nil) + + req := ctrl.Request{NamespacedName: types.NamespacedName{Name: coalescedKey}} + + // First call: should run reconcileAll and succeed. + if _, err := r.Reconcile(context.Background(), req); err != nil { + t.Fatalf("first Reconcile returned error: %v", err) + } + + // Second call immediately after: should return early without running reconcileAll. + result, err := r.Reconcile(context.Background(), req) + if err != nil { + t.Fatalf("second Reconcile returned error: %v", err) + } + if result.RequeueAfter == 0 { + t.Fatal("second Reconcile: expected non-zero RequeueAfter (min interval early return), got 0") + } + if result.RequeueAfter > minInterval { + t.Errorf("second Reconcile: RequeueAfter %v exceeds MinReconcileInterval %v", result.RequeueAfter, minInterval) + } + + // Only one CRD should exist — the second call must not have triggered reconcileAll. + var list v1alpha1.FlavorGroupCapacityList + if err := fakeClient.List(context.Background(), &list); err != nil { + t.Fatalf("failed to list CRDs: %v", err) + } + if len(list.Items) != 1 { + t.Errorf("expected 1 CRD (only first reconcile ran), got %d", len(list.Items)) + } +} From 77f7ed16b99c8454f0cc9df2a8b8e5b050483187 Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Wed, 8 Jul 2026 16:47:58 +0200 Subject: [PATCH 3/8] fix --- internal/scheduling/reservations/capacity/controller.go | 6 ------ 1 file changed, 6 deletions(-) diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index 8bd4ee01d..411365f84 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -9,7 +9,6 @@ import ( "hash/fnv" "sort" "strings" - "sync" "time" hv1 "github.com/cobaltcore-dev/openstack-hypervisor-operator/api/v1" @@ -48,7 +47,6 @@ type Reconciler struct { schedulerClient *reservations.SchedulerClient config Config - mu sync.Mutex lastReconcileAt time.Time } @@ -65,10 +63,8 @@ func NewController(c client.Client, config Config, vmSource reservations.VMSourc // watched CRD changes, and also on the periodic RequeueAfter floor set by ReconcileInterval. // If called sooner than MinReconcileInterval since the last successful run, it returns early. func (c *Reconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result, error) { - c.mu.Lock() elapsed := time.Since(c.lastReconcileAt) minInterval := c.config.MinReconcileInterval.Duration - c.mu.Unlock() if c.lastReconcileAt != (time.Time{}) && elapsed < minInterval { remaining := minInterval - elapsed @@ -84,9 +80,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result return ctrl.Result{}, err } - c.mu.Lock() c.lastReconcileAt = time.Now() - c.mu.Unlock() return ctrl.Result{RequeueAfter: c.config.ReconcileInterval.Duration}, nil } From d10682554fbd5189c161cd9e24810713237ad3b8 Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Mon, 13 Jul 2026 13:43:40 +0200 Subject: [PATCH 4/8] validate reconcileinterval --- cmd/manager/main.go | 4 ++++ internal/scheduling/reservations/capacity/config.go | 10 ++++++++++ 2 files changed, 14 insertions(+) diff --git a/cmd/manager/main.go b/cmd/manager/main.go index 2ffeb0f55..d8508d1c7 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -765,6 +765,10 @@ func main() { setupLog.Info("enabling controller", "controller", "capacity-controller") capacityConfig := conf.GetConfigOrDie[capacity.Config]() capacityConfig.ApplyDefaults() + if err := capacityConfig.Validate(); err != nil { + setupLog.Error(err, "invalid capacity-controller config") + os.Exit(1) + } capacityMonitor := capacity.NewMonitor(multiclusterClient) if err := metrics.Registry.Register(&capacityMonitor); err != nil { diff --git a/internal/scheduling/reservations/capacity/config.go b/internal/scheduling/reservations/capacity/config.go index cd33875b9..c99346ac7 100644 --- a/internal/scheduling/reservations/capacity/config.go +++ b/internal/scheduling/reservations/capacity/config.go @@ -4,6 +4,7 @@ package capacity import ( + "fmt" "time" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" @@ -53,6 +54,15 @@ func (c *Config) ApplyDefaults() { } } +// Validate checks that the config is internally consistent after defaults are applied. +func (c *Config) Validate() error { + if c.ReconcileInterval.Duration <= c.MinReconcileInterval.Duration { + return fmt.Errorf("capacityReconcileInterval (%s) must be greater than capacityMinReconcileInterval (%s)", + c.ReconcileInterval.Duration, c.MinReconcileInterval.Duration) + } + return nil +} + func DefaultConfig() Config { return Config{ ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, From f9ceedeba577ad8e13bcaa5d60c51d461959737f Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Mon, 13 Jul 2026 13:53:04 +0200 Subject: [PATCH 5/8] watch predicates --- .../reservations/capacity/controller.go | 63 +++++++++++++++++-- 1 file changed, 57 insertions(+), 6 deletions(-) diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index 411365f84..92c20550e 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -20,7 +20,9 @@ import ( ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller" + "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" schedulerapi "github.com/cobaltcore-dev/cortex/api/external/nova" @@ -34,10 +36,59 @@ import ( var log = ctrl.Log.WithName("capacity-controller").WithValues("module", "capacity") // coalescedKey is the single reconcile key used for all watch events. -// All CRD changes (Knowledge, Hypervisor, CommittedResource, Pipeline) are coalesced -// into this one key so rapid changes never cause rapid-fire scheduler probes. +// All CRD changes are coalesced into this one key so rapid changes never +// cause rapid-fire scheduler probes. const coalescedKey = "capacity" +// flavorGroupsKnowledgePredicate fires only for the "flavor_groups" Knowledge object. +// Other Knowledge objects (different extractors) are irrelevant to capacity. +var flavorGroupsKnowledgePredicate = predicate.NewPredicateFuncs(func(obj client.Object) bool { + k, ok := obj.(*v1alpha1.Knowledge) + if !ok { + return false + } + return k.Spec.Extractor.Name == "flavor_groups" +}) + +// hvCapacityChangePredicate fires only when Allocation, EffectiveCapacity, or Capacity +// changed on a Hypervisor, and only for hypervisors that carry an AZ label. +// Label/annotation-only updates and unrelated status field changes are ignored. +var hvCapacityChangePredicate = predicate.Funcs{ + CreateFunc: func(e event.CreateEvent) bool { + _, hasAZ := e.Object.GetLabels()["topology.kubernetes.io/zone"] + return hasAZ + }, + UpdateFunc: func(e event.UpdateEvent) bool { + if _, hasAZ := e.ObjectNew.GetLabels()["topology.kubernetes.io/zone"]; !hasAZ { + return false + } + oldHV, ok1 := e.ObjectOld.(*hv1.Hypervisor) + newHV, ok2 := e.ObjectNew.(*hv1.Hypervisor) + if !ok1 || !ok2 { + return false + } + return !capacityMapsEqual(oldHV.Status.Allocation, newHV.Status.Allocation) || + !capacityMapsEqual(oldHV.Status.EffectiveCapacity, newHV.Status.EffectiveCapacity) || + !capacityMapsEqual(oldHV.Status.Capacity, newHV.Status.Capacity) + }, + DeleteFunc: func(e event.DeleteEvent) bool { return true }, + GenericFunc: func(e event.GenericEvent) bool { return false }, +} + +// capacityMapsEqual returns true if two resource maps are equal by value. +func capacityMapsEqual(a, b map[hv1.ResourceName]resource.Quantity) bool { + if len(a) != len(b) { + return false + } + for k, va := range a { + vb, ok := b[k] + if !ok || va.Cmp(vb) != 0 { + return false + } + } + return true +} + // Reconciler reconciles FlavorGroupCapacity CRDs, driven by both watch events and a // periodic floor timer. All watch events are coalesced into a single reconcile key so // rapid changes produce at most one queued reconcile. @@ -99,17 +150,17 @@ func (c *Reconciler) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client bldr := multicluster.BuildController(mcl, mgr) var err error - bldr, err = bldr.WatchesMulticluster(&v1alpha1.Knowledge{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + bldr, err = bldr.WatchesMulticluster(&v1alpha1.Knowledge{}, handler.EnqueueRequestsFromMapFunc(coalesce), flavorGroupsKnowledgePredicate) if err != nil { return fmt.Errorf("failed to watch Knowledge: %w", err) } - bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{}, handler.EnqueueRequestsFromMapFunc(coalesce), hvCapacityChangePredicate) if err != nil { return fmt.Errorf("failed to watch Hypervisor: %w", err) } - bldr, err = bldr.WatchesMulticluster(&v1alpha1.CommittedResource{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + bldr, err = bldr.WatchesMulticluster(&v1alpha1.Reservation{}, handler.EnqueueRequestsFromMapFunc(coalesce)) if err != nil { - return fmt.Errorf("failed to watch CommittedResource: %w", err) + return fmt.Errorf("failed to watch Reservation: %w", err) } bldr, err = bldr.WatchesMulticluster(&v1alpha1.Pipeline{}, handler.EnqueueRequestsFromMapFunc(coalesce)) if err != nil { From bd2e5b20197d677b8fce9b59a34c1ce168513751 Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Mon, 13 Jul 2026 13:59:53 +0200 Subject: [PATCH 6/8] add helm values --- helm/bundles/cortex-nova/values.yaml | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index 1171f3b01..3a64ff0ce 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -157,8 +157,11 @@ cortex-scheduling-controllers: capacityTotalPipeline: "kvm-report-capacity" # Pipeline used for the current-state capacity probe (considers current VM allocations). capacityPlaceablePipeline: "kvm-general-purpose-load-balancing-no-history" - # How often the capacity controller re-runs its scheduler probes. + # How often the capacity reconciler re-runs its scheduler probes (periodic floor). capacityReconcileInterval: 5m + # Minimum time between two consecutive capacity reconcile runs. + # Prevents back-to-back reconciles on rapid watch events. + capacityMinReconcileInterval: 30s # If true, the external scheduler API will limit the list of hosts in its # response to those included in the scheduling request. novaLimitHostsToRequest: true From 8fac933812083706fd02453a3bcd30bd417b8bdd Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Mon, 13 Jul 2026 14:02:16 +0200 Subject: [PATCH 7/8] fix --- internal/scheduling/reservations/capacity/controller.go | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index 92c20550e..9200b986b 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -71,7 +71,10 @@ var hvCapacityChangePredicate = predicate.Funcs{ !capacityMapsEqual(oldHV.Status.EffectiveCapacity, newHV.Status.EffectiveCapacity) || !capacityMapsEqual(oldHV.Status.Capacity, newHV.Status.Capacity) }, - DeleteFunc: func(e event.DeleteEvent) bool { return true }, + DeleteFunc: func(e event.DeleteEvent) bool { + _, hasAZ := e.Object.GetLabels()["topology.kubernetes.io/zone"] + return hasAZ + }, GenericFunc: func(e event.GenericEvent) bool { return false }, } @@ -117,7 +120,7 @@ func (c *Reconciler) Reconcile(ctx context.Context, _ ctrl.Request) (ctrl.Result elapsed := time.Since(c.lastReconcileAt) minInterval := c.config.MinReconcileInterval.Duration - if c.lastReconcileAt != (time.Time{}) && elapsed < minInterval { + if !c.lastReconcileAt.IsZero() && elapsed < minInterval { remaining := minInterval - elapsed LoggerFromContext(ctx).V(1).Info("skipping reconcile: min interval not elapsed", "elapsed", elapsed.Round(time.Second), From f9c35593340a6500a50437dfab3f7861832c4f22 Mon Sep 17 00:00:00 2001 From: Julius Clausnitzer Date: Mon, 13 Jul 2026 16:10:35 +0200 Subject: [PATCH 8/8] add edge case --- internal/scheduling/reservations/capacity/controller.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index acc16b7b1..d3f5c8249 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -59,9 +59,14 @@ var hvCapacityChangePredicate = predicate.Funcs{ return hasAZ }, UpdateFunc: func(e event.UpdateEvent) bool { - if _, hasAZ := e.ObjectNew.GetLabels()["topology.kubernetes.io/zone"]; !hasAZ { + oldAZ := e.ObjectOld.GetLabels()["topology.kubernetes.io/zone"] + newAZ := e.ObjectNew.GetLabels()["topology.kubernetes.io/zone"] + if newAZ == "" { return false } + if oldAZ != newAZ { + return true + } oldHV, ok1 := e.ObjectOld.(*hv1.Hypervisor) newHV, ok2 := e.ObjectNew.(*hv1.Hypervisor) if !ok1 || !ok2 {