diff --git a/cmd/manager/main.go b/cmd/manager/main.go index af86e3d84..d8508d1c7 100644 --- a/cmd/manager/main.go +++ b/cmd/manager/main.go @@ -765,22 +765,25 @@ 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 { 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/helm/bundles/cortex-nova/values.yaml b/helm/bundles/cortex-nova/values.yaml index d931d22f6..27acd10e1 100644 --- a/helm/bundles/cortex-nova/values.yaml +++ b/helm/bundles/cortex-nova/values.yaml @@ -157,8 +157,11 @@ cortex-scheduling-controllers: capacityTotalPipeline: "kvm-general-purpose-load-balancing" # Pipeline used for the current-state capacity probe (considers current VM allocations). capacityPlaceablePipeline: "kvm-general-purpose-load-balancing" - # 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 diff --git a/internal/scheduling/reservations/capacity/config.go b/internal/scheduling/reservations/capacity/config.go index d4834c53e..4b9292042 100644 --- a/internal/scheduling/reservations/capacity/config.go +++ b/internal/scheduling/reservations/capacity/config.go @@ -4,16 +4,24 @@ package capacity import ( + "fmt" "time" 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 +40,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 } @@ -43,11 +54,21 @@ 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}, - TotalPipeline: "kvm-general-purpose-load-balancing", - PlaceablePipeline: "kvm-general-purpose-load-balancing", - SchedulerURL: "http://localhost:8080/scheduler/nova/external", + ReconcileInterval: metav1.Duration{Duration: 5 * time.Minute}, + MinReconcileInterval: metav1.Duration{Duration: 30 * time.Second}, + TotalPipeline: "kvm-general-purpose-load-balancing", + PlaceablePipeline: "kvm-general-purpose-load-balancing", + SchedulerURL: "http://localhost:8080/scheduler/nova/external", } } diff --git a/internal/scheduling/reservations/capacity/controller.go b/internal/scheduling/reservations/capacity/controller.go index 257afe79e..d3f5c8249 100644 --- a/internal/scheduling/reservations/capacity/controller.go +++ b/internal/scheduling/reservations/capacity/controller.go @@ -19,28 +19,98 @@ 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/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" "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 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 { + 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 { + 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 { + _, hasAZ := e.Object.GetLabels()["topology.kubernetes.io/zone"] + return hasAZ + }, + 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. +type Reconciler struct { client client.Client vmSource reservations.VMSource schedulerClient *reservations.SchedulerClient config Config + + 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 +118,66 @@ 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) { + elapsed := time.Since(c.lastReconcileAt) + minInterval := c.config.MinReconcileInterval.Duration + + 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), + "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.lastReconcileAt = time.Now() + + 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), flavorGroupsKnowledgePredicate) + if err != nil { + return fmt.Errorf("failed to watch Knowledge: %w", err) } + 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.Reservation{}, handler.EnqueueRequestsFromMapFunc(coalesce)) + if err != nil { + return fmt.Errorf("failed to watch Reservation: %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 +192,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 +244,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 +346,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 +527,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 +620,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, @@ -599,7 +712,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) @@ -635,7 +748,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) diff --git a/internal/scheduling/reservations/capacity/controller_test.go b/internal/scheduling/reservations/capacity/controller_test.go index 7d066bee6..0ba4b8f5c 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) } @@ -881,3 +883,116 @@ func TestProbeScheduler_SetsCapacityProbeIntent(t *testing.T) { t.Errorf("capacity probe must send ProjectID cortex-capacity-probe, got %q", capturedReq.Spec.Data.ProjectID) } } + +// 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)) + } +}