feat: client cache for configurable list of gvks

cmd/manager/main.go

@@ -64,6 +64,7 @@ import (
 	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/quota"
+	"github.com/cobaltcore-dev/cortex/pkg/clientcache"
 	"github.com/cobaltcore-dev/cortex/pkg/conf"
 	"github.com/cobaltcore-dev/cortex/pkg/monitoring"
 	"github.com/cobaltcore-dev/cortex/pkg/multicluster"
@@ -393,6 +394,22 @@ func main() {
 		os.Exit(1)
 	}
 
+	// Transparent in-process overlay cache for CRDs that are eventually
+	// consistent across in-pod clients (e.g. Reservations). Writes populate an
+	// overlay; reads merge it with the informer result until the real object is
+	// observed. *multicluster.Client serves as both the inner client.Client and
+	// the InformerSource; the cache itself has no multicluster dependency.
+	clientCacheConfig := conf.GetConfigOrDie[clientcache.RootConfig]()
+	cachingClient, err := clientcache.New(multiclusterClient, multiclusterClient, scheme, clientCacheConfig.ClientCache)
+	if err != nil {
+		setupLog.Error(err, "unable to create client cache")
+		os.Exit(1)
+	}
+	if err := mgr.Add(cachingClient); err != nil {
+		setupLog.Error(err, "unable to add client cache to manager")
+		os.Exit(1)
+	}
+
 	// Our custom monitoring registry can add prometheus labels to all metrics.
 	// This is useful to distinguish metrics from different deployments.
 	metricsConfig := conf.GetConfigOrDie[monitoring.Config]()
@@ -424,10 +441,10 @@ func main() {
 	commitmentsConfig := conf.GetConfigOrDie[commitments.Config]()
 	var commitmentsVMSource reservations.VMSource
 	if commitmentsConfig.DatasourceName != "" {
-		commitmentsVMSource = reservations.NewPostgresVMSource(multiclusterClient, commitmentsConfig.DatasourceName)
+		commitmentsVMSource = reservations.NewPostgresVMSource(cachingClient, commitmentsConfig.DatasourceName)
 	}
 	if slices.Contains(mainConfig.EnabledControllers, "committed-resource-reservations-controller") {
-		commitmentsAPI := commitmentsapi.NewAPIWithConfig(multiclusterClient, commitmentsConfig.API, commitmentsVMSource)
+		commitmentsAPI := commitmentsapi.NewAPIWithConfig(cachingClient, commitmentsConfig.API, commitmentsVMSource)
 		commitmentsAPI.Init(mux, metrics.Registry, ctrl.Log.WithName("commitments-api"))
 	}
 
@@ -447,8 +464,8 @@ func main() {
 		metrics.Registry.MustRegister(noHostFoundCounter)
 		metrics.Registry.MustRegister(placementCounter)
 		// Inferred through the base controller.
-		filterWeigherController.Client = multiclusterClient
-		filterWeigherController.CRRecorder.Client = multiclusterClient
+		filterWeigherController.Client = cachingClient
+		filterWeigherController.CRRecorder.Client = cachingClient
 		if err := filterWeigherController.SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "nova FilterWeigherPipelineController")
 			os.Exit(1)
@@ -463,7 +480,7 @@ func main() {
 		novaClient := nova.NewNovaClient()
 		novaClientConfig := conf.GetConfigOrDie[nova.NovaClientConfig]()
 		if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
-			return novaClient.Init(ctx, multiclusterClient, novaClientConfig)
+			return novaClient.Init(ctx, cachingClient, novaClientConfig)
 		})); err != nil {
 			setupLog.Error(err, "unable to initialize nova client")
 			os.Exit(1)
@@ -474,15 +491,15 @@ func main() {
 			Breaker: &nova.DetectorCycleBreaker{NovaClient: novaClient},
 		}
 		// Inferred through the base controller.
-		deschedulingsController.Client = multiclusterClient
+		deschedulingsController.Client = cachingClient
 		if err := (deschedulingsController).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "nova DetectorPipelineController")
 			os.Exit(1)
 		}
 		go deschedulingsController.CreateDeschedulingsPeriodically(ctx)
 		// Deschedulings cleanup on startup
 		if err := (&nova.DeschedulingsCleanup{
-			Client: multiclusterClient,
+			Client: cachingClient,
 			Scheme: mgr.GetScheme(),
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "Cleanup")
@@ -502,13 +519,13 @@ func main() {
 		novaClient := nova.NewNovaClient()
 		novaClientConfig := conf.GetConfigOrDie[nova.NovaClientConfig]()
 		if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
-			return novaClient.Init(ctx, multiclusterClient, novaClientConfig)
+			return novaClient.Init(ctx, cachingClient, novaClientConfig)
 		})); err != nil {
 			setupLog.Error(err, "unable to initialize nova client")
 			os.Exit(1)
 		}
 		if err := (&nova.DeschedulingsExecutor{
-			Client:     multiclusterClient,
+			Client:     cachingClient,
 			Scheme:     mgr.GetScheme(),
 			Conf:       executorConfig,
 			NovaClient: novaClient,
@@ -519,8 +536,8 @@ func main() {
 	}
 	if slices.Contains(mainConfig.EnabledControllers, "hypervisor-overcommit-controller") {
 		hypervisorOvercommitController := &nova.HypervisorOvercommitController{}
-		hypervisorOvercommitController.Client = multiclusterClient
-		if err := hypervisorOvercommitController.SetupWithManager(mgr); err != nil {
+		hypervisorOvercommitController.Client = cachingClient
+		if err := hypervisorOvercommitController.SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller",
 				"controller", "HypervisorOvercommitController")
 			os.Exit(1)
@@ -532,7 +549,7 @@ func main() {
 			Monitor: filterWeigherPipelineMonitor,
 		}
 		// Inferred through the base controller.
-		controller.Client = multiclusterClient
+		controller.Client = cachingClient
 		if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler")
 			os.Exit(1)
@@ -552,7 +569,7 @@ func main() {
 			Monitor: filterWeigherPipelineMonitor,
 		}
 		// Inferred through the base controller.
-		controller.Client = multiclusterClient
+		controller.Client = cachingClient
 		if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler")
 			os.Exit(1)
@@ -572,7 +589,7 @@ func main() {
 			Monitor: filterWeigherPipelineMonitor,
 		}
 		// Inferred through the base controller.
-		controller.Client = multiclusterClient
+		controller.Client = cachingClient
 		if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler")
 			os.Exit(1)
@@ -591,7 +608,7 @@ func main() {
 			Monitor: filterWeigherPipelineMonitor,
 		}
 		// Inferred through the base controller.
-		controller.Client = multiclusterClient
+		controller.Client = cachingClient
 		if err := (controller).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "DecisionReconciler")
 			os.Exit(1)
@@ -607,11 +624,11 @@ func main() {
 
 	if slices.Contains(mainConfig.EnabledControllers, "committed-resource-reservations-controller") {
 		setupLog.Info("enabling controller", "controller", "committed-resource-reservations-controller")
-		monitor := reservations.NewMonitor(multiclusterClient)
+		monitor := reservations.NewMonitor(cachingClient)
 		metrics.Registry.MustRegister(&monitor)
 
 		if err := (&commitments.CommitmentReservationController{
-			Client: multiclusterClient,
+			Client: cachingClient,
 			Scheme: mgr.GetScheme(),
 			Conf:   commitmentsConfig.ReservationController,
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
@@ -621,11 +638,11 @@ func main() {
 
 		crControllerConf := commitmentsConfig.CommittedResourceController
 
-		crControllerMonitor := commitments.NewCRControllerMonitor(multiclusterClient)
+		crControllerMonitor := commitments.NewCRControllerMonitor(cachingClient)
 		metrics.Registry.MustRegister(&crControllerMonitor)
 
 		if err := (&commitments.CommittedResourceController{
-			Client:   multiclusterClient,
+			Client:   cachingClient,
 			Scheme:   mgr.GetScheme(),
 			Conf:     crControllerConf,
 			Monitor:  &crControllerMonitor,
@@ -643,7 +660,7 @@ func main() {
 			usageReconcilerConf := commitmentsConfig.UsageReconciler
 			usageReconcilerConf.ApplyDefaults()
 			if err := (&commitments.UsageReconciler{
-				Client:   multiclusterClient,
+				Client:   cachingClient,
 				Conf:     usageReconcilerConf,
 				VMSource: commitmentsVMSource,
 				Monitor:  usageReconcilerMonitor,
@@ -658,15 +675,15 @@ func main() {
 		monitor := datasources.NewMonitor()
 		metrics.Registry.MustRegister(&monitor)
 		if err := (&openstack.OpenStackDatasourceReconciler{
-			Client:  multiclusterClient,
+			Client:  cachingClient,
 			Scheme:  mgr.GetScheme(),
 			Monitor: monitor,
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "OpenStackDatasourceReconciler")
 			os.Exit(1)
 		}
 		if err := (&prometheus.PrometheusDatasourceReconciler{
-			Client:  multiclusterClient,
+			Client:  cachingClient,
 			Scheme:  mgr.GetScheme(),
 			Monitor: monitor,
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
@@ -679,7 +696,7 @@ func main() {
 		monitor := extractor.NewMonitor()
 		metrics.Registry.MustRegister(&monitor)
 		if err := (&extractor.KnowledgeReconciler{
-			Client:  multiclusterClient,
+			Client:  cachingClient,
 			Scheme:  mgr.GetScheme(),
 			Monitor: monitor,
 			Conf:    conf.GetConfigOrDie[extractor.KnowledgeReconcilerConfig](),
@@ -688,7 +705,7 @@ func main() {
 			os.Exit(1)
 		}
 		if err := (&extractor.TriggerReconciler{
-			Client: multiclusterClient,
+			Client: cachingClient,
 			Scheme: mgr.GetScheme(),
 			Conf:   conf.GetConfigOrDie[extractor.TriggerReconcilerConfig](),
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
@@ -700,7 +717,7 @@ func main() {
 		setupLog.Info("enabling controller", "controller", "kpis-controller")
 		kpisControllerConfig := conf.GetConfigOrDie[kpis.ControllerConfig]()
 		if err := (&kpis.Controller{
-			Client: multiclusterClient,
+			Client: cachingClient,
 			Config: kpisControllerConfig,
 		}).SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "KPIController")
@@ -732,7 +749,7 @@ func main() {
 		if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
 			// Create PostgresReader from the configured Datasource CRD
 			// This runs after the cache is started
-			postgresReader, err := external.NewPostgresReader(ctx, multiclusterClient, failoverConfig.DatasourceName)
+			postgresReader, err := external.NewPostgresReader(ctx, cachingClient, failoverConfig.DatasourceName)
 			if err != nil {
 				setupLog.Error(err, "unable to create postgres reader for failover controller",
 					"datasourceName", failoverConfig.DatasourceName)
@@ -748,7 +765,7 @@ func main() {
 			// 1. Watch-based per-reservation reconciliation (acknowledgment, validation)
 			// 2. Periodic bulk VM processing (creating/assigning reservations)
 			failoverController := failover.NewFailoverReservationController(
-				multiclusterClient,
+				cachingClient,
 				vmSource,
 				failoverConfig,
 				schedulerClient,
@@ -791,12 +808,12 @@ func main() {
 			os.Exit(1)
 		}
 
-		capacityMonitor := capacity.NewMonitor(multiclusterClient)
+		capacityMonitor := capacity.NewMonitor(cachingClient)
 		if err := metrics.Registry.Register(&capacityMonitor); err != nil {
 			setupLog.Error(err, "failed to register capacity monitor metrics, continuing without metrics")
 		}
 
-		if err := capacity.NewController(multiclusterClient, capacityConfig, commitmentsVMSource).
+		if err := capacity.NewController(cachingClient, capacityConfig, commitmentsVMSource).
 			SetupWithManager(mgr, multiclusterClient); err != nil {
 			setupLog.Error(err, "unable to create controller", "controller", "capacity")
 			os.Exit(1)
@@ -828,7 +845,7 @@ func main() {
 		// Defer initialization until the manager starts (cache must be ready for postgres reader)
 		if err := mgr.Add(manager.RunnableFunc(func(ctx context.Context) error {
 			// Create PostgresReader from the configured Datasource CRD
-			postgresReader, err := external.NewPostgresReader(ctx, multiclusterClient, datasourceName)
+			postgresReader, err := external.NewPostgresReader(ctx, cachingClient, datasourceName)
 			if err != nil {
 				setupLog.Error(err, "unable to create postgres reader for quota controller",
 					"datasourceName", datasourceName)
@@ -841,7 +858,7 @@ func main() {
 
 			// Create the quota controller
 			quotaController := quota.NewQuotaController(
-				multiclusterClient,
+				cachingClient,
 				vmSource,
 				quotaConfig,
 				quotaMetrics,
@@ -903,11 +920,11 @@ func main() {
 		setupLog.Info("starting commitments syncer")
 		syncerMonitor := commitments.NewSyncerMonitor()
 		must.Succeed(metrics.Registry.Register(syncerMonitor))
-		syncer := commitments.NewSyncer(multiclusterClient, syncerMonitor)
+		syncer := commitments.NewSyncer(cachingClient, syncerMonitor)
 		syncerConfig := conf.GetConfigOrDie[commitments.SyncerConfig]()
 		syncerConfig.FlavorGroupResourceConfig = commitmentsConfig.API.FlavorGroupResourceConfig
 		if err := (&task.Runner{
-			Client:   multiclusterClient,
+			Client:   cachingClient,
 			Interval: syncerConfig.SyncInterval.Duration,
 			Name:     "commitments-sync-task",
 			Run:      func(ctx context.Context) error { return syncer.SyncReservations(ctx) },
@@ -921,11 +938,11 @@ func main() {
 		setupLog.Info("starting nova history cleanup task")
 		historyCleanupConfig := conf.GetConfigOrDie[nova.HistoryCleanupConfig]()
 		if err := (&task.Runner{
-			Client:   multiclusterClient,
+			Client:   cachingClient,
 			Interval: time.Hour,
 			Name:     "nova-history-cleanup-task",
 			Run: func(ctx context.Context) error {
-				return nova.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig)
+				return nova.HistoryCleanup(ctx, cachingClient, historyCleanupConfig)
 			},
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to add nova history cleanup task to manager")
@@ -936,11 +953,11 @@ func main() {
 		setupLog.Info("starting manila history cleanup task")
 		historyCleanupConfig := conf.GetConfigOrDie[manila.HistoryCleanupConfig]()
 		if err := (&task.Runner{
-			Client:   multiclusterClient,
+			Client:   cachingClient,
 			Interval: time.Hour,
 			Name:     "manila-history-cleanup-task",
 			Run: func(ctx context.Context) error {
-				return manila.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig)
+				return manila.HistoryCleanup(ctx, cachingClient, historyCleanupConfig)
 			},
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to add manila history cleanup task to manager")
@@ -951,11 +968,11 @@ func main() {
 		setupLog.Info("starting cinder history cleanup task")
 		historyCleanupConfig := conf.GetConfigOrDie[cinder.HistoryCleanupConfig]()
 		if err := (&task.Runner{
-			Client:   multiclusterClient,
+			Client:   cachingClient,
 			Interval: time.Hour,
 			Name:     "cinder-history-cleanup-task",
 			Run: func(ctx context.Context) error {
-				return cinder.HistoryCleanup(ctx, multiclusterClient, historyCleanupConfig)
+				return cinder.HistoryCleanup(ctx, cachingClient, historyCleanupConfig)
 			},
 		}).SetupWithManager(mgr); err != nil {
 			setupLog.Error(err, "unable to add cinder history cleanup task to manager")

helm/bundles/cortex-nova/values.yaml

@@ -110,6 +110,9 @@ cortex: &cortex
           - kvm.cloud.sap/v1/Hypervisor
           - kvm.cloud.sap/v1/HypervisorList
           - v1/Secret
+    clientcache:
+      gvks:
+        - cortex.cloud/v1alpha1/Reservation
     keystoneSecretRef:
       name: cortex-nova-openstack-keystone
       namespace: default

internal/scheduling/nova/hypervisor_overcommit_controller.go

@@ -217,7 +217,7 @@ func (c *HypervisorOvercommitController) predicateRemoteHypervisor() predicate.P
 // SetupWithManager sets up the controller with the Manager and a multicluster
 // client. The multicluster client is used to watch for changes in the
 // Hypervisor CRD across all clusters and trigger reconciliations accordingly.
-func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager) (err error) {
+func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager, mcl *multicluster.Client) (err error) {
 	// This will load the config in a safe way and gracefully handle errors.
 	c.config, err = conf.GetConfig[HypervisorOvercommitConfig]()
 	if err != nil {
@@ -227,12 +227,6 @@ func (c *HypervisorOvercommitController) SetupWithManager(mgr ctrl.Manager) (err
 	if err := c.config.Validate(); err != nil {
 		return err
 	}
-	// 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 hypervisor crd may be distributed across multiple remote clusters.
 	bldr, err = bldr.WatchesMulticluster(&hv1.Hypervisor{},

internal/scheduling/nova/hypervisor_overcommit_controller_test.go

@@ -725,7 +725,7 @@ func TestHypervisorOvercommitController_SetupWithManager_InvalidClient(t *testin
 	// SetupWithManager should fail - either because config loading fails
 	// (in test environment without config files) or because the client
 	// is not a multicluster client.
-	err := controller.SetupWithManager(mgr)
+	err := controller.SetupWithManager(mgr, nil)
 	if err == nil {
 		t.Error("expected error when calling SetupWithManager, got nil")
 	}